vbhanuchander-lang commented on PR #309:
URL:
https://github.com/apache/flink-connector-kafka/pull/309#issuecomment-5837270571
@MartijnVisser #300 landed on the 21st, so this is rebased onto current
`main` (`9c69ad6`) and reduced exactly as you asked: the set, its initialiser
and the `addReader` reset are gone, and what is left is the one
`readersWithNoMoreSplits.clear()` in `onHandleSubscribedStreamsFetch`. 2 files,
+90/-4, no conflicts. No need to apologise for the overlap — #300 is the better
base for the dedup anyway, since the deferral belongs with it.
I confirmed your reason for wanting #300 first rather than taking it on
trust. Four of its five bounded-source tests do fail on the old branch, and the
cause is the one you gave: the first signal leaves before the other clusters
have assigned, so with the set making it the only signal the reader finishes
holding one cluster's splits. Those five tests are in the class I ran below and
all pass with this change on top.
**One correction, because it changes what the test looks like rather than
whether it exists.** You said #300's wiring of `signalNoMoreSplitsCallback` in
`TestKafkaEnumContextProxyFactory` makes the test I could not write work on
that base. The wiring does make `handleNoMoreSplits` reachable from the proxy,
but that was not what blocked me, and a test written the obvious way still
fails on this base. I instrumented `handleNoMoreSplits` to find out why:
```
after the metadata change, adding cluster 1 to a stream already on cluster 0:
contexts=[kafka-cluster-0, kafka-cluster-1]
noMoreSplitsFlags=[kafka-cluster-0=false, kafka-cluster-1=true]
alreadySignalled=[] <- the clear works
```
The set is clear, the readers are registered, and the signal still does not
go out, because `kafka-cluster-0` never re-signals. A **retained** cluster is
recreated from state in which its partitions are already assigned, so
`KafkaSourceEnumerator#checkPartitionChanges` hits `partitionChange.isEmpty()`
and returns before it would reach `signalNoMoreSplits`. Its new proxy's
`isNoMoreSplits()` stays `false`, `allEnumeratorsHaveSignalledNoMoreSplits` is
never true, and the "not ready" branch is taken no matter what the set
contains. That is the FLINK-31006 gap, and it sits one level below the callback.
What makes it testable is choosing a metadata change with **no retained
cluster**: a switchover, where the stream moves off cluster 0 and onto cluster
1. Every sub enumerator is then new, every one of them signals, the
precondition is reachable, and the clear is exactly what decides the outcome:
- on `main`: `Expected size: 2 but was: 1` — the readers are never told again
- with the clear: passes
So `testBoundedSourceCompletesReadersAgainAfterMetadataChange` is a real
behavioural test, not a `Whitebox` assertion on the set, and the coverage
question from my earlier comment is closed. The comment in the test records why
a retained cluster cannot be used, so nobody rewrites it into the version that
cannot pass.
Verified on the rebased tree: `DynamicKafkaSourceEnumeratorTest` 36/36,
`DynamicKafkaSourceEnumeratorRecoveryTest` 10/10,
`DynamicKafkaSourceReaderTest` 15/15 — 61 tests, 0 failures. `spotless:check`
and `checkstyle:check` clean on JDK 17, confirmed by checking the goals
actually ran rather than the exit code.
Two notes for the re-scope you mentioned: the title and description now
describe only the clear, and the FLINK-40586 summary still describes the
re-signal behaviour that #300 fixed, so it will need rewording to match
whenever you get to it.
--
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]
