syhily commented on code in PR #21252:
URL: https://github.com/apache/flink/pull/21252#discussion_r1019857265
##########
flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/function/ControlSource.java:
##########
@@ -183,37 +191,58 @@ public List<String> getExpectedRecords() {
private static class StopSignal implements Closeable {
private static final Logger LOG =
LoggerFactory.getLogger(StopSignal.class);
- private final String topic;
private final int desiredCounts;
// This is a thread-safe list.
private final List<String> consumedRecords;
private final AtomicLong deadline;
private final ExecutorService executor;
+ private final Consumer<String> consumer;
+ private final AtomicReference<PulsarClientException>
throwableException;
public StopSignal(
PulsarRuntimeOperator operator, String topic, int
messageCounts, Duration timeout) {
- this.topic = topic;
this.desiredCounts = messageCounts;
this.consumedRecords = Collections.synchronizedList(new
ArrayList<>(messageCounts));
this.deadline = new AtomicLong(timeout.toMillis() +
System.currentTimeMillis());
this.executor = Executors.newSingleThreadExecutor();
+ ConsumerBuilder<String> consumerBuilder =
+ operator.client()
+ .newConsumer(Schema.STRING)
+ .topic(topic)
+ .subscriptionName(randomAlphanumeric(10))
+ .subscriptionMode(Durable)
+ .subscriptionType(Exclusive)
+
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
+ this.consumer = sneakyClient(consumerBuilder::subscribe);
+ this.throwableException = new AtomicReference<>();
// Start consuming.
executor.execute(
() -> {
while (consumedRecords.size() < desiredCounts) {
// This method would block until we consumed a
message.
int counts = desiredCounts -
consumedRecords.size();
- List<Message<String>> messages =
- operator.receiveMessages(this.topic,
Schema.STRING, counts);
- for (Message<String> message : messages) {
- consumedRecords.add(message.getValue());
+ for (int i = 0; i < counts; i++) {
+ try {
+ Message<String> message =
consumer.receive();
+ consumedRecords.add(message.getValue());
+ } catch (PulsarClientException e) {
+ throwableException.set(e);
+ break;
+ }
}
}
});
}
public boolean canStop() {
+ PulsarClientException exception = throwableException.get();
+ if (exception != null) {
+ LOG.error("Error in consuming messages from Pulsar.");
+ LOG.error("", exception);
+ return true;
+ }
Review Comment:
Yep. I think you are right.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
