dalelane commented on code in PR #293:
URL:
https://github.com/apache/flink-connector-kafka/pull/293#discussion_r4056599002
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java:
##########
@@ -266,6 +277,456 @@ void testOffsetCommitOnCheckpointComplete() throws
Exception {
}
}
+ /** Writes the records that a {@link #offsetConvergenceScenarios} scenario
needs. */
+ @FunctionalInterface
+ private interface RecordProducer {
+ void produce(String topic) throws Throwable;
+ }
+
+ private static Stream<Arguments> offsetConvergenceScenarios() {
Review Comment:
My initial reaction was that I would like to keep all seven tests. They were
written with a view to describing scenarios that felt likely to regress when
this area of code is rewritten or refactored in future - even if today they go
through the same code.
That duration is still hard to defend, and is a bad smell, so I spent a bit
of time tracing through to work out why these tests were taking so long.
The delay was coming from
`KafkaSourceFetcherManager.enqueueOffsetsCommitTask` which was enqueuing the
commit task. It introduced a poll.timeout.ms delay per commit whenever the
split is idle, because the fetcher was blocked in the consumer.poll. Across the
test scenarios the 10 secs every time added up to a very slow batch of tests.
I think this was highlighting a real issue, not just a test problem. Looking
at other places where tasks are enqueued onto live fetchers, this is paired
with a wakeup. I've added a wakeup call in
e44287bedc002e0b88379beba44fcac7afbdf4ab and that does bring the time for
KafkaSourceReaderTest back down to ~49s for me, which feels like a more
reasonable increase.
--
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]
