tzulitai commented on code in PR #46:
URL:
https://github.com/apache/flink-connector-kafka/pull/46#discussion_r1316526004
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java:
##########
@@ -452,12 +453,9 @@ public void onCompletion(RecordMetadata metadata,
Exception exception) {
asyncProducerException = decorateException(metadata,
exception, producer);
}
+ // Checking for exceptions from previous writes
mailboxExecutor.submit(
- () -> {
- // Checking for exceptions from previous writes
- checkAsyncException();
- },
- "Update error metric");
+ KafkaWriter.this::checkAsyncException, "Update error
metric");
Review Comment:
minor nit:
was this necessary? Perhaps could have been a separate `[hotfix]` commit.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducerITCase.java:
##########
@@ -173,9 +172,10 @@ private static Properties getProperties() {
}
private static List<Consumer<FlinkKafkaInternalProducer<?, ?>>>
provideTransactionsFinalizer() {
- return Lists.newArrayList(
- FlinkKafkaInternalProducer::commitTransaction,
- FlinkKafkaInternalProducer::abortTransaction);
+ List<Consumer<FlinkKafkaInternalProducer<?, ?>>> ret = new
ArrayList<>();
+ ret.add(FlinkKafkaInternalProducer::commitTransaction);
+ ret.add(FlinkKafkaInternalProducer::abortTransaction);
Review Comment:
nit:
a lot of these can probably be replaced with:
```
Arrays.asList(
item1,
item2
)
```
to remove the Java boilerplates.
##########
flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/shuffle/KafkaShuffleITCase.java:
##########
@@ -374,17 +371,16 @@ private void testRecordSerDe(TimeCharacteristic
timeCharacteristic) throws Excep
// Records in a single partition are kept in order
Collection<ConsumerRecord<byte[], byte[]>> records =
- Iterables.getOnlyElement(
- testKafkaShuffleProducer(
- topic(
- "test_serde-" +
UUID.randomUUID(),
- timeCharacteristic),
- env,
- 1,
- 1,
- numElementsPerProducer,
- timeCharacteristic)
- .values());
+ testKafkaShuffleProducer(
+ topic("test_serde-" + UUID.randomUUID(),
timeCharacteristic),
+ env,
+ 1,
+ 1,
+ numElementsPerProducer,
+ timeCharacteristic)
+ .values()
+ .iterator()
+ .next();
Review Comment:
(same for some other places where `Iterables.getOnlyElement` was replaced)
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java:
##########
@@ -245,12 +245,12 @@ public void close() throws Exception {
LOG.debug("Closing writer with {}", currentProducer);
closeAll(
this::abortCurrentProducer,
- closer,
producerPool::clear,
() -> {
checkState(currentProducer.isClosed());
currentProducer = null;
});
+ closeAll(closeables);
Review Comment:
originally, with a single `closeAll` we had an implicit ordering on the
sequence of what stuff to close after the other.
Now with the two `closeAll`s, the ordering has changed.
Could you double check if moving the closeables to the end would affect
anything?
I checked and I think it is fine, but could use a second pair of eyes.
##########
flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/shuffle/KafkaShuffleITCase.java:
##########
@@ -374,17 +371,16 @@ private void testRecordSerDe(TimeCharacteristic
timeCharacteristic) throws Excep
// Records in a single partition are kept in order
Collection<ConsumerRecord<byte[], byte[]>> records =
- Iterables.getOnlyElement(
- testKafkaShuffleProducer(
- topic(
- "test_serde-" +
UUID.randomUUID(),
- timeCharacteristic),
- env,
- 1,
- 1,
- numElementsPerProducer,
- timeCharacteristic)
- .values());
+ testKafkaShuffleProducer(
+ topic("test_serde-" + UUID.randomUUID(),
timeCharacteristic),
+ env,
+ 1,
+ 1,
+ numElementsPerProducer,
+ timeCharacteristic)
+ .values()
+ .iterator()
+ .next();
Review Comment:
Moving away from `Iterables.getOnlyElement`, you'd lose the test coverage
where we test there's actually only 1 element in the iterable. Can you add an
additional assertion where there is no more next element, to retain that
coverage?
--
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]
