[
https://issues.apache.org/jira/browse/FLINK-38617?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18076253#comment-18076253
]
Hang JI edited comment on FLINK-38617 at 4/25/26 6:28 PM:
----------------------------------------------------------
Hi, I have also encountered the same issue and spent some time digging into it.
What I found is:
h3. {color:#172b4d}Root Cause{color}
{color:#172b4d}The race lives between the async offset commit and the Kafka
source reader shutdown when {{stop-with-savepoint}} is invoked. As described in
the ticket.{color}
{color:#172b4d}In fact{color}
# {color:#172b4d}First, Flink stops a job, {color}
# {color:#172b4d}Then the savepoint completes on the JM,
{{KafkaSourceReader.notifyCheckpointComplete arrives on the mailbox thread,
which enqueues a commitAsync}} task onto the {{SplitFetcher's taskQueue}} and
returns.{color}
# {color:#172b4d}After that, {{streamTask.closeAllOperators}} runs:
{{SourceReader.close}} calls {{{}SplitFetcher.shutdown{}}}, the queue is
discarded along with the still-pending commit, and the fetcher thread is never
scheduled in between. (All of it inside the same millisecond).{color}
The async hop isn't optional. KafkaConsumer is not thread-safe, and
{{notifyCheckpointComplete}} runs on the mailbox thread while the consumer
lives on the fetcher thread, so the commit has to be enqueued cross-thread. I
checked some other implementations of SourceReader (e.g.,
[{color:#0747a6}jdbcSourceReader{color}|https://github.com/apache/flink-connector-jdbc/blob/main/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/reader/JdbcSourceReader.java],
[MongoSourceReader|https://github.com/apache/flink-connector-mongodb/blob/main/flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/source/reader/MongoSourceReader.java],
{color:#0747a6}KinesisStreamsSourceReader{color}, etc.); most of them either
don't override {{notifyCheckpointComplete}} or implement it synchronously,
which is why this only shows up on Kafka.
FLINK-23408 intended to order {{notifyCheckpointComplete}} before close on
{{{}stop-with-savepoint{}}}, because "{_}Before finishing a task we should wait
for a checkpoint issued after {{finish()}} to commit all pending transactions
created from the {{finish()}} method.{_}" The implementation waits for the
call to return, but not for the side effects it enqueued.
{color:#172b4d}The reason why Flink close can't tell the two paths apart is
that {{SourceReader}} extends {{{}AutoCloseable{}}}, and
{{AutoCloseable.close()}} has no parameters. {color}
{color:#172b4d}{{Stop-with-savepoint}} and cancel both fall into the same
{{SplitFetcher.shutdown}} that discards the queue, which is fine for {{{}Flink
cancel{}}}, but wrong for {{Flink stop}} (with Kafka source).{color}
h3. Impact
- For sinks that commit synchronously on checkpoint (e.g.,
{color:#172b4d}FileSink with OnCheckpointRollingPolicy{color}), duplicates show
up when {{stop-with-savepoint}} is followed by a redeploy that loses the source
state: files are already on the filesystem, but the offsets the savepoint was
supposed to commit never made it to Kafka. ( my case )
- For real-time sinks (e.g., Elasticsearch / Kafka / FileSink
withDefaultRollingPolicy), every {{stop-with-savepoint}} can produce duplicates
regardless of state, because the data is already downstream before the dropped
commit would have mattered.
h3. Proposed fix
Add a graceful close path on {{{}SourceReader{}}}:
{code:java}
@PublicEvolving
default void closeGracefully() throws Exception {
close();
}
{code}
Default fallback to {{close()}} keeps existing implementations working.
On {{{}stop-with-savepoint{}}}, the call chain becomes:
{code:java}
SourceOperator.close()
└─ if (!isCanceled()) sourceReader.closeGracefully()
└─ SourceReaderBase.closeGracefully()
└─ SplitFetcherManager.closeGracefully()
└─ SplitFetcher.shutdownGracefully()
// sets draining=true, drains taskQueue before exit
{code}
On cancel, {{SourceOperator.close()}} keeps calling {{close()}} directly - same
path as today, queue discarded.
Readers based on {{SourceReaderBase}} (Kafka included) pick this up
automatically; readers that implement {{SourceReader}} directly need to
override the new method to opt in.
I looked at a few alternatives — a flag on {{{}SourceReaderContext{}}},
blocking inside {{{}notifyCheckpointComplete{}}}, hooking into
{{{}StreamOperator.finish(){}}}, but each one breaks something (lifecycle scope
of the context, mailbox non-blocking, two-phase-commit ordering for the
{{finish()}} option). Adding a method ended up being the cleanest of the bunch.
h3. Verification
Reproduced on *{color:#0747a6}1.20.3{color}* with {{stop-with-savepoint}}
(Kafka Source + FileSink + OnCheckpointRollingPolicy and Kafka Source +
KafkaSink), where the source state is lost on restart, downstream consistently
sees duplicate records.
With the patch, the same scenario runs clean across repeated runs.
h3. Open question
I'm not sure if this needs a {*}FLIP{*}. This fix adds a {{@PublicEvolving
default}} method on a {{@Public}} interface, which is the same shape as
FLIP-217 / {{{}pauseOrResumeSplits{}}}.
If the direction looks right, happy to take this on as my first contribution to
Flink
was (Author: JIRAUSER313047):
Hi, I have also encountered the same issue and spent some time digging into it.
What I found is:
h3. {color:#172b4d}Root Cause{color}
{color:#172b4d}The race lives between the async offset commit and the Kafka
source reader shutdown when {{stop-with-savepoint}} is invoked. As described in
the ticket.{color}
{color:#172b4d}In fact{color}
# {color:#172b4d}First, Flink stops a job, {color}
# {color:#172b4d}Then the savepoint completes on the JM,
{{KafkaSourceReader.notifyCheckpointComplete arrives on the mailbox thread,
which enqueues a commitAsync}} task onto the {{SplitFetcher's taskQueue}} and
returns.{color}
# {color:#172b4d}After that, {{streamTask.closeAllOperators}} runs:
{{SourceReader.close}} calls {{{}SplitFetcher.shutdown{}}}, the queue is
discarded along with the still-pending commit, and the fetcher thread is never
scheduled in between. (All of it inside the same millisecond).{color}
The async hop isn't optional. KafkaConsumer is not thread-safe, and
{{notifyCheckpointComplete}} runs on the mailbox thread while the consumer
lives on the fetcher thread, so the commit has to be enqueued cross-thread. I
checked some other implementations of SourceReader (e.g.,
[{color:#0747a6}jdbcSourceReader{color}|https://github.com/apache/flink-connector-jdbc/blob/main/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/reader/JdbcSourceReader.java],
[MongoSourceReader|https://github.com/apache/flink-connector-mongodb/blob/main/flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/source/reader/MongoSourceReader.java],
{color:#0747a6}KinesisStreamsSourceReader{color}, etc.); most of them either
don't override {{notifyCheckpointComplete}} or implement it synchronously,
which is why this only shows up on Kafka.
FLINK-23408 intended to order {{notifyCheckpointComplete}} before close on
{{{}stop-with-savepoint{}}}, because "{_}Before finishing a task we should wait
for a checkpoint issued after {{finish()}} to commit all pending transactions
created from the {{finish()}} method.{_}" The implementation waits for the
call to return, but not for the side effects it enqueued.
{color:#172b4d}The reason why Flink close can't tell the two paths apart is
that {{SourceReader}} extends {{{}AutoCloseable{}}}, and
{{AutoCloseable.close()}} has no parameters. {color}
{color:#172b4d}{{Stop-with-savepoint}} and cancel both fall into the same
{{SplitFetcher.shutdown}} that discards the queue, which is fine for {{{}Flink
cancel{}}}, but wrong for {{Flink stop}} (with Kafka source).{color}
h3. Impact
- For sinks that commit synchronously on checkpoint (e.g.,
{color:#172b4d}FileSink with OnCheckpointRollingPolicy{color}), duplicates show
up when {{stop-with-savepoint}} is followed by a redeploy that loses the source
state: files are already on the filesystem, but the offsets the savepoint was
supposed to commit never made it to Kafka. ( my case )
- For real-time sinks (e.g., Elasticsearch / Kafka / FileSink
withDefaultRollingPolicy), every \{{stop-with-savepoint }}can produce
duplicates regardless of state, because the data is already downstream before
the dropped commit would have mattered.
h3. Proposed fix
Add a graceful close path on {{{}SourceReader{}}}:
{code:java}
@PublicEvolving
default void closeGracefully() throws Exception {
close();
}
{code}
Default fallback to {{close()}} keeps existing implementations working.
On {{{}stop-with-savepoint{}}}, the call chain becomes:
{code:java}
SourceOperator.close()
└─ if (!isCanceled()) sourceReader.closeGracefully()
└─ SourceReaderBase.closeGracefully()
└─ SplitFetcherManager.closeGracefully()
└─ SplitFetcher.shutdownGracefully()
// sets draining=true, drains taskQueue before exit
{code}
On cancel, {{SourceOperator.close()}} keeps calling {{close()}} directly - same
path as today, queue discarded.
Readers based on {{SourceReaderBase}} (Kafka included) pick this up
automatically; readers that implement {{SourceReader}} directly need to
override the new method to opt in.
I looked at a few alternatives — a flag on {{{}SourceReaderContext{}}},
blocking inside {{{}notifyCheckpointComplete{}}}, hooking into
{{{}StreamOperator.finish(){}}}, but each one breaks something (lifecycle scope
of the context, mailbox non-blocking, two-phase-commit ordering for the
{{finish()}} option). Adding a method ended up being the cleanest of the bunch.
h3. Verification
Reproduced on *{color:#0747a6}1.20.3{color}* with {{stop-with-savepoint}}
(Kafka Source + FileSink + OnCheckpointRollingPolicy and Kafka Source +
KafkaSink), where the source state is lost on restart, downstream consistently
sees duplicate records.
With the patch, the same scenario runs clean across repeated runs.
h3. Open question
I'm not sure if this needs a {*}FLIP{*}. This fix adds a {{@PublicEvolving
default}} method on a {{@Public}} interface, which is the same shape as
FLIP-217 / {{{}pauseOrResumeSplits{}}}.
If the direction looks right, happy to take this on as my first contribution to
Flink
> Kafka consumer do not commit offset when stop-with-savepoint
> ------------------------------------------------------------
>
> Key: FLINK-38617
> URL: https://issues.apache.org/jira/browse/FLINK-38617
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Common
> Affects Versions: 1.17.2, 1.20.3
> Environment: flink 1.17
> Reporter: xingsuo-zbz
> Priority: Critical
>
> Recently, we saw Kafka consumer do not commit offset when stop-with-savepoint.
> 1. When a checkpoint completed, CheckpointCoordinator call
> `KafkaSourceReader.notifyCheckpointComplete()`.
> 2. Then `splitFetcherManager.commitOffsets()` enqueue an OffsetsCommitTask to
> `SplitFetcher.taskQueue`.
> 3. Before the taskQueue is emptied, the loop of `SplitFetcher.runOnce()` is
> interrupted by `SplitFetcher.shutdown()`.
>
> During the "stop-with-savepoint" period, using Arthas to view the call logs
> also confirmed this issue：
> {panel:title=Arthas log}
> [arthas@1]$ watch
> org.apache.flink.connector.kafka.source.reader.KafkaPartitionSplitReader
> notifyCheckpointComplete -x 2 -n 200
> Press Q or Ctrl+C to abort.
> Affect(class count: 2 , method count: 1) cost in 518 ms, listenerId: 1
> session (396ccb27-c9b4-438c-b137-6936c7756765) is closed because server is
> going to shutdown.
> $ command terminated with exit code 137
> {panel}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
