MartijnVisser commented on code in PR #313:
URL:
https://github.com/apache/flink-connector-kafka/pull/313#discussion_r3981062947
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/testutils/KafkaUtil.java:
##########
@@ -55,6 +55,8 @@ public class KafkaUtil {
private static final Logger LOG = LoggerFactory.getLogger(KafkaUtil.class);
private static final Duration CONSUMER_POLL_DURATION =
Duration.ofSeconds(1);
+ private static final Duration OPEN_TRANSACTION_SETTLE_TIMEOUT =
Duration.ofSeconds(10);
Review Comment:
The drains that assert while transactions are open pay this in full:
`testAbortOnClose` 6.7s -> 16.6s,
`testAbortTransactionsAfterScaleInBeforeFirstCheckpoint` 56.2s -> 96.2s, so
+52.7s per cell across the two classes, times three JDK cells on PR CI and
seven on the weekly. Please add a variant of `drainAllRecordsFromTopic` that
skips the wait when transactions are expected to be open.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/testutils/KafkaUtil.java:
##########
@@ -192,6 +195,55 @@ public static List<ConsumerRecord<byte[], byte[]>>
drainAllRecordsFromTopic(
}
}
+ /**
+ * Returns the end offsets up to which {@code consumer} drains. Under
{@code read_committed},
+ * {@link KafkaConsumer#endOffsets} is the last stable offset, which
trails the high watermark
+ * until the broker has written the markers of a transaction it has
already acknowledged as
+ * committed. Waits a bounded time for the two to meet so that a commit
that completed a moment
+ * ago is not cut off. If a transaction stays open, returns the last
stable offset as is and
+ * logs which offsets are pinned, so a truncated drain names its cause.
+ */
+ private static Map<TopicPartition, Long> getSettledEndOffsets(
+ String topic,
+ KafkaConsumer<byte[], byte[]> consumer,
+ Properties consumerConfig,
+ Set<TopicPartition> topicPartitions) {
+ Map<TopicPartition, Long> endOffsets =
consumer.endOffsets(topicPartitions);
+ if (!"read_committed"
+
.equals(consumerConfig.getProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG))) {
Review Comment:
Nit: `getProperty` returns `null` for a non-`String` value, and callers here
already mix the two (`enable.auto.commit` is stored as a `Boolean`). Works
today because only the three-arg overload writes `isolation.level`;
`String.valueOf(consumerConfig.get(...))` makes it safe.
--
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]
