MartijnVisser commented on code in PR #313:
URL:
https://github.com/apache/flink-connector-kafka/pull/313#discussion_r3982365424
##########
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:
Filter on transaction state instead. A commit in flight is `PrepareCommit`,
a transaction the test left open is `Ongoing`, and `filterStates` is
broker-side, so the wait never sees the second. On a prototype
`testAbortOnClose` and `testAbortTransactionsAfterScaleInBeforeFirstCheckpoint`
are back to their base times.
##########
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:
This is worse than a nit. The commit marker takes an offset, so when it's
the marker that's pending the watermark reads X and the LSO settles at X+1, and
`endOffsets.equals(highWatermarks)` never holds. The wait then always runs the
full 10 s and warns about open transactions while printing an LSO above the
watermark.
--
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]
