syhily commented on code in PR #21252:
URL: https://github.com/apache/flink/pull/21252#discussion_r1019826795
##########
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<>();
Review Comment:
I think it's ok to use `CompletableFuture` with better readability.
--
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
