[
https://issues.apache.org/jira/browse/FLINK-4027?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15333502#comment-15333502
]
ASF GitHub Bot commented on FLINK-4027:
---------------------------------------
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/2108#discussion_r67318807
--- Diff:
flink-streaming-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerBase.java
---
@@ -276,6 +315,38 @@ public void close() throws Exception {
checkErroneous();
}
+ // ------------------- Logic for handling checkpoint flushing
-------------------------- //
+
+ private void acknowledgeMessage() {
+ if(!flushOnCheckpoint) {
+ // the logic is disabled
+ return;
+ }
+ pendingRecords--;
+ }
+
+ protected abstract void flush();
+
+ @Override
+ public Serializable snapshotState(long checkpointId, long
checkpointTimestamp) {
+ if(flushOnCheckpoint) {
+ // flushing is activated: We need to wait until
pendingRecords is 0
+ flush();
+
+ if(pendingRecords != 0) {
--- End diff --
space
> FlinkKafkaProducer09 sink can lose messages
> -------------------------------------------
>
> Key: FLINK-4027
> URL: https://issues.apache.org/jira/browse/FLINK-4027
> Project: Flink
> Issue Type: Bug
> Components: Kafka Connector
> Affects Versions: 1.0.3
> Reporter: Elias Levy
> Assignee: Robert Metzger
> Priority: Critical
>
> The FlinkKafkaProducer09 sink appears to not offer at-least-once guarantees.
> The producer is publishing messages asynchronously. A callback can record
> publishing errors, which will be raised when detected. But as far as I can
> tell, there is no barrier to wait for async errors from the sink when
> checkpointing or to track the event time of acked messages to inform the
> checkpointing process.
> If a checkpoint occurs while there are pending publish requests, and the
> requests return a failure after the checkpoint occurred, those message will
> be lost as the checkpoint will consider them processed by the sink.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
