AHeise commented on a change in pull request #17019:
URL: https://github.com/apache/flink/pull/17019#discussion_r698535910
##########
File path:
flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
##########
@@ -164,13 +163,12 @@ public void write(IN element, Context context) throws
IOException {
@Override
public List<KafkaCommittable> prepareCommit(boolean flush) {
flushRecords(flush);
- List<KafkaCommittable> committables = precommit();
- currentProducer = createProducer();
- return committables;
+ return precommit();
}
@Override
- public List<KafkaWriterState> snapshotState() throws IOException {
+ public List<KafkaWriterState> snapshotState(long checkpointId) throws
IOException {
+ currentProducer = createProducer(checkpointId);
Review comment:
I now explicitly nulled the `currentProducer`. That still leaves room in
the future for errors but at least not correctness issues.
--
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]
