Gerrrr commented on code in PR #46:
URL:
https://github.com/apache/flink-connector-kafka/pull/46#discussion_r1317650559
##########
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:
Thank you for catching this! Turns out, integration test failures were due
to the incorrect ordering here. For example, `KafkaSinkITCase` failed with
```
2023-09-06T06:59:29.8283020Z java.lang.IllegalStateException: null
2023-09-06T06:59:29.8283703Z at
org.apache.flink.util.Preconditions.checkState(Preconditions.java:177)
~[flink-core-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8284351Z at
org.apache.flink.connector.kafka.sink.KafkaWriter.lambda$close$0(KafkaWriter.java:250)
~[classes/:?]
2023-09-06T06:59:29.8285042Z at
org.apache.flink.util.IOUtils.closeAll(IOUtils.java:255)
~[flink-core-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8285660Z at
org.apache.flink.util.IOUtils.closeAll(IOUtils.java:234)
~[flink-core-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8286306Z at
org.apache.flink.util.IOUtils.closeAll(IOUtils.java:223)
~[flink-core-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8286870Z at
org.apache.flink.connector.kafka.sink.KafkaWriter.close(KafkaWriter.java:246)
~[classes/:?]
2023-09-06T06:59:29.8287538Z at
org.apache.flink.util.IOUtils.closeAll(IOUtils.java:255)
~[flink-core-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8288146Z at
org.apache.flink.util.IOUtils.closeAll(IOUtils.java:234)
~[flink-core-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8288751Z at
org.apache.flink.util.IOUtils.closeAll(IOUtils.java:223)
~[flink-core-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8290025Z at
org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperator.close(SinkWriterOperator.java:216)
~[flink-streaming-java-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8311878Z at
org.apache.flink.streaming.runtime.tasks.StreamOperatorWrapper.close(StreamOperatorWrapper.java:163)
~[flink-streaming-java-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8313214Z at
org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.closeAllOperators(RegularOperatorChain.java:125)
~[flink-streaming-java-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8314283Z at
org.apache.flink.streaming.runtime.tasks.StreamTask.closeAllOperators(StreamTask.java:1043)
~[flink-streaming-java-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8315237Z at
org.apache.flink.streaming.runtime.tasks.StreamTask.afterInvoke(StreamTask.java:911)
~[flink-streaming-java-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8316141Z at
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:794)
~[flink-streaming-java-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8317027Z at
org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:952)
~[flink-runtime-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8317875Z at
org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:931)
~[flink-runtime-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8318834Z at
org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:745)
~[flink-runtime-1.17.0.jar:1.17.0]
2023-09-06T06:59:29.8319526Z at
org.apache.flink.runtime.taskmanager.Task.run(Task.java:562)
~[flink-runtime-1.17.0.jar:1.17.0]
```
--
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]
