You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/11/09 10:51:59 UTC
[GitHub] [flink] XComp commented on a diff in pull request #21252: [FLINK-29830][Connector/Pulsar] Create the topic with schema before consuming messages in PulsarSinkITCase
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org