[
https://issues.apache.org/jira/browse/FLINK-31006?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18113926#comment-18113926
]
Martijn Visser commented on FLINK-31006:
----------------------------------------
The bounded FileSource and NumberSequenceSource do not carry this state at all.
Their enumerators
are pull-based: {{StaticFileSplitEnumerator.addReader}} does nothing, and
{{handleSplitRequest}}
either assigns the next split or calls {{signalNoMoreSplits}} for that subtask,
recomputed from the
assigner on every request. There is nothing to lose across a restore, so the
first split request
after recovery answers correctly. {{IteratorSourceEnumerator}} is the same
shape and says so in
{{addReader}}: it deliberately assigns nothing on registration and waits for
the reader's explicit
request.
The Pulsar source is the closer comparison, because it is push-based like
Kafka. It does not have
this bug, and the reason is precisely what this fix changes.
{{SplitAssignerImpl.registerTopicPartitions}} sets {{initialized = true}}
unconditionally at the end
of every discovery, whether or not the fetched set contained anything new, and
{{PulsarSourceEnumerator}} calls it on every discovery with no emptiness guard.
Its
{{noMoreSplits(reader)}} is then derived as
{{!enablePartitionDiscovery && initialized &&
!pendingPartitionSplits.containsKey(reader)}}. So the
nearest neighbour of the Kafka enumerator already does what this fix makes
Kafka do. That also
confirms the 2023 assessment that the Pulsar component on this ticket can be
dropped, from the code
rather than from the {{StopCursor}} argument given at the time.
The Kafka source pushes splits eagerly too, and its {{handleSplitRequest}} is
empty. Because nothing
ever asks, the enumerator has to decide for itself when the input is finished,
and it remembers that
in {{noMoreNewPartitionSplits}}, which is derived state that never enters the
checkpoint. That is
the bug, and it is specific to push-based enumerators.
That argues against snapshotting {{noMoreSplitsAssignment}} in
{{SourceReaderBase}}: pull-based
sources need no reader-side memory, and reader state is redistributed on a
rescale, so the flag
would follow the wrong subtask.
It also rules out the cheapest enumerator-side variant, which is to derive the
flag from the
{{initialDiscoveryFinished}} value that is already checkpointed and signal
straight from
{{addReader}}. Measured on a restored bounded enumerator whose state holds one
topic while a second
topic appeared while the job was down, with the reader registering before the
post-restore
discovery completes, the event order for that reader is:
{noformat}
[signalNoMoreSplits:0, assignSplits:0, signalNoMoreSplits:0]
{noformat}
The reader is told the input ended and is assigned a partition afterwards. That
is what
apache/flink#21909 was rejected for. The same variant also leaves the
fresh-start half of the bug
unfixed, because {{initialDiscoveryFinished}} is false there.
Waiting for the discovery callback and signalling from
{{handlePartitionSplitChanges}} keeps the
assignment before the signal in both registration orders and needs no state
format change.
## Evidence behind it
- `StaticFileSplitEnumerator` and `IteratorSourceEnumerator` read at
`~/Developer/flink`.
- Pulsar read from `apache/flink-connector-pulsar` on GitHub:
`SplitAssignerImpl.java:77-79`
(unconditional `initialized = true`), `:117-121` (derived `noMoreSplits`),
`:124-126`
(`initialized` is not snapshotted), and `PulsarSourceEnumerator.java:211-212`
(unguarded call).
- Ordering measured with a recording `MockSplitEnumeratorContext` subclass; test
`testRestoredBoundedEnumeratorAssignsBeforeItSignals`, which passes on the
committed fix and
fails under the derived-flag variant with the trace above.
- Under the derived-flag variant, `KafkaSourceEnumeratorTest` is 26 tests with
2 failures: the
ordering test and `testBoundedSourceWithoutPartitionsSignalsNoMoreSplits`. It
does pass the
headline restore test, which is why the shortcut looks attractive.
> job is not finished when using pipeline mode to run bounded source like kafka
> -----------------------------------------------------------------------------
>
> Key: FLINK-31006
> URL: https://issues.apache.org/jira/browse/FLINK-31006
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
> Affects Versions: 2.1.0
> Reporter: Jacky Lau
> Assignee: Martijn Visser
> Priority: Major
> Labels: pull-request-available
> Fix For: kafka-5.1.0, kafka-4.0.2, kafka-5.0.1, kafka-3.4.1
>
> Attachments: image-2023-02-10-13-20-52-890.png,
> image-2023-02-10-13-23-38-430.png, image-2023-02-10-13-24-46-929.png,
> image-2023-03-04-01-04-18-658.png, image-2023-03-04-01-05-25-335.png,
> image-2023-03-04-01-07-04-927.png, image-2023-03-04-01-07-36-168.png,
> image-2023-03-04-01-08-29-042.png, image-2023-03-04-01-09-24-199.png
>
>
> when i do failover works like kill jm/tm when using pipeline mode to run
> bounded source like kafka, i found job is not finished, when every partition
> data has consumed.
>
> After dig into code, i found this logical not run when JM recover. the
> partition infos are not changed. so noMoreNewPartitionSplits is not set to
> true. then this will not run
>
> !image-2023-02-10-13-23-38-430.png!
>
> !image-2023-02-10-13-24-46-929.png!
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
