MartijnVisser commented on code in PR #313:
URL:
https://github.com/apache/flink-connector-kafka/pull/313#discussion_r3981062952
##########
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 value stored as a non-`String`, and
the callers here already mix the two
(`KafkaSinkITCase.getKafkaClientConfiguration` stores `enable.auto.commit` as a
`Boolean`). Only the three-arg overload writes `isolation.level` today so it
works, but a caller of the two-arg `drainAllRecordsFromTopic` that sets it
itself would silently keep the old behaviour.
`String.valueOf(consumerConfig.get(...))` is enough.
##########
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))) {
+ return endOffsets;
+ }
+ final Properties uncommittedConfig = new Properties();
+ uncommittedConfig.putAll(consumerConfig);
+ uncommittedConfig.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG,
"read_uncommitted");
+ try (KafkaConsumer<byte[], byte[]> uncommittedConsumer =
+ new KafkaConsumer<>(uncommittedConfig)) {
+ final Map<TopicPartition, Long> highWatermarks =
Review Comment:
Nit: the high watermarks are read once and the loop compares a moving last
stable offset against that fixed target. Correct here because the drain runs
after the job finished, but worth a line in the javadoc since the method reads
as reusable.
##########
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:
There's no way to opt out of this, so the drains that assert *while*
transactions are open pay it in full. Measured on this branch against
`beb52b43`:
- `ExactlyOnceKafkaWriterITCase#testAbortOnClose` 6.7s -> 16.6s
- `KafkaSinkITCase#testAbortTransactionsAfterScaleInBeforeFirstCheckpoint`
56.2s -> 96.2s (4 parameter sets, one timeout each)
At class level that is +52.7s across `KafkaSinkITCase` and
`ExactlyOnceKafkaWriterITCase`, so those two tests are the whole cost. Times
three JDK cells on PR CI and seven on the weekly.
Please give `drainAllRecordsFromTopic` a variant that says transactions are
expected to be open and skips the wait, and keep the 10 s for the drains that
assert a settled topic.
--
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]
