Efrat19 commented on code in PR #292:
URL:
https://github.com/apache/flink-connector-kafka/pull/292#discussion_r3764180777
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderTest.java:
##########
@@ -377,6 +379,58 @@ void testPauseOrResumeSplitsWithUnassignedPartition() {
Collections.singletonList(unassignedSplit));
}
+ @Test
+ public void testDefaultPollTimeoutIsTenSeconds() {
+ // The default is part of the connector's behavior contract, changing
it would silently
+ // change how long the split reader blocks on idle partitions for all
existing jobs.
+ assertThat(KafkaSourceOptions.POLL_TIMEOUT_MS.defaultValue())
+ .isEqualTo(Duration.ofSeconds(10).toMillis());
+ }
+
+ @Test
+ public void testConfiguredPollTimeoutIsUsedForPolling() throws Exception {
+ final Properties props = new Properties();
+ props.setProperty(KafkaSourceOptions.POLL_TIMEOUT_MS.key(), "500");
+ KafkaPartitionSplitReader reader =
+ createReader(props,
UnregisteredMetricsGroup.createSourceReaderMetricGroup());
+
+ // TOPIC3 is empty, so fetching from it blocks until the poll timeout
expires
+ reader.handleSplitsChanges(
+ new SplitsAddition<>(
+ Collections.singletonList(
+ new KafkaPartitionSplit(
+ new TopicPartition(TOPIC3, 0),
+
KafkaPartitionSplit.EARLIEST_OFFSET))));
+ // Warm up the consumer so that metadata fetching is not part of the
measurement below
+ reader.fetch();
+
+ final long startTime = System.nanoTime();
+ RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>> records =
reader.fetch();
+ final Duration fetchDuration = Duration.ofNanos(System.nanoTime() -
startTime);
+
+ assertThat(records.nextSplit()).isNull();
+ // Generously below the 10 seconds default, but way above the
configured 500ms, so that the
+ // assertion only fails if the configured timeout is ignored
+ assertThat(fetchDuration).isLessThan(Duration.ofSeconds(5));
Review Comment:
Why 5 seconds and not the configured 500ms?
Also maybe its enough to have a visibleForTesting
KafkaPartitionSplitReader#getPollTimeout() instead of testing kafka client
behavior here, making the test lighter and more deterministic, wdyt?
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReaderTest.java:
##########
@@ -377,6 +379,58 @@ void testPauseOrResumeSplitsWithUnassignedPartition() {
Collections.singletonList(unassignedSplit));
}
+ @Test
+ public void testDefaultPollTimeoutIsTenSeconds() {
Review Comment:
IMO asserting the default value is used when no other is provided would make
this test more valueable, wdyt?
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceOptions.java:
##########
@@ -58,6 +58,18 @@ public class KafkaSourceOptions {
.defaultValue(true)
.withDescription("Whether to commit consuming offset on
checkpoint.");
+ public static final ConfigOption<Long> POLL_TIMEOUT_MS =
+ ConfigOptions.key("poll.timeout.ms")
+ .longType()
+ .defaultValue(Duration.ofSeconds(10).toMillis())
+ .withDescription(
Review Comment:
It might be important to also mention that poll returns immediately if there
are records available.
([docs](https://kafka.apache.org/25/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#poll-java.time.Duration-))
--
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]
