XComp commented on code in PR #21252:
URL: https://github.com/apache/flink/pull/21252#discussion_r1016457270
##########
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 =
Review Comment:
```suggestion
final ConsumerBuilder<String> consumerBuilder =
```
nit
##########
flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntimeOperator.java:
##########
@@ -180,6 +180,10 @@ public void createTopic(String topic, int
numberOfPartitions) {
}
}
+ public void createSchema(String topic, Schema<?> schema) {
Review Comment:
```suggestion
@VisibleForTesting
public void createSchema(String topic, Schema<?> schema) {
```
##########
flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntimeOperator.java:
##########
@@ -180,6 +180,10 @@ public void createTopic(String topic, int
numberOfPartitions) {
}
}
+ public void createSchema(String topic, Schema<?> schema) {
+ sneakyAdmin(() -> admin().schemas().createSchema(topic,
schema.getSchemaInfo()));
Review Comment:
Why do we use these `sneaky` utility functions everywhere? It feels a bit
fishy to hide checked exceptions. Either we have to document the reason
properly or we should get rid of it and do proper exception handling. WDYT?
I understand that such a change would be out-of-scope of this PR. We would
need to create a follow-up ticket.
##########
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:
What about working with `CompletableFutures` here? It feels more natural to
use rather than utilizing an `AtomicReference` here.
##########
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:
I feel like there is a non-optimal separation of concerns here: we shouldn't
log in a method that checks the state but rather have the logging be done in
the calling method.
##########
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)
Review Comment:
```suggestion
.subscriptionMode(SubscriptionMode.Durable)
.subscriptionType(SubscriptionType.Exclusive)
```
nit: The enum usage here is not obvious because the enum identifiers are not
uppercased. Could we replace the static import for these with a qualified
access to improve readability?
##########
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;
Review Comment:
The Consumer is only accessed in the constructor. I don't see a necessity to
create a dedicated field for that one. Or am I missing something in this
aspect? :thinking:
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
