You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ar...@apache.org on 2021/10/28 06:39:51 UTC
[flink] branch release-1.14 updated: [FLINK-23391][connector/kafka]
Fix flaky Kafka source metric test by retrying notifyCheckpointComplete
until success or timeout
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.14 by this push:
new d58469d [FLINK-23391][connector/kafka] Fix flaky Kafka source metric test by retrying notifyCheckpointComplete until success or timeout
d58469d is described below
commit d58469d6eed64a5b5cc7abbc36c641667f1f9201
Author: Qingsheng Ren <re...@gmail.com>
AuthorDate: Tue Oct 19 16:36:26 2021 +0800
[FLINK-23391][connector/kafka] Fix flaky Kafka source metric test by retrying notifyCheckpointComplete until success or timeout
When calling KafkaSourceReader.notifyCheckpointComplete(), KafkaConsumer.commitAsync() might fail because of coordinator movement or network instability during the test. Retrying offset commit until success will increase the stability of the case and has no side effect since this operation is idempotent.
---
.../kafka/source/reader/KafkaSourceReaderTest.java | 26 ++++++++++++++++------
1 file changed, 19 insertions(+), 7 deletions(-)
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java
index 5911031..e671520 100644
--- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReaderTest.java
@@ -329,28 +329,40 @@ public class KafkaSourceReaderTest extends SourceReaderTestBase<KafkaPartitionSp
assertEquals(INITIAL_OFFSET, getCommittedOffsetMetric(tp1, metricListener));
// Trigger offset commit
- reader.snapshotState(15213L);
- reader.notifyCheckpointComplete(15213L);
+ final long checkpointId = 15213L;
+ reader.snapshotState(checkpointId);
waitUtil(
- () -> reader.getOffsetsToCommit().isEmpty(),
+ () -> {
+ try {
+ reader.notifyCheckpointComplete(checkpointId);
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to notify checkpoint complete to reader", e);
+ }
+ return reader.getOffsetsToCommit().isEmpty();
+ },
Duration.ofSeconds(60),
+ Duration.ofSeconds(1),
String.format(
"Offsets are not committed successfully. Dangling offsets: %s",
reader.getOffsetsToCommit()));
- // Metric "commit-total" of KafkaConsumer should be 1
- assertEquals(1, getKafkaConsumerMetric("commit-total", metricListener));
+ // Metric "commit-total" of KafkaConsumer should be greater than 0
+ // It's hard to know the exactly number of commit because of the retry
+ MatcherAssert.assertThat(
+ getKafkaConsumerMetric("commit-total", metricListener),
+ Matchers.greaterThan(0L));
// Committed offset should be NUM_RECORD_PER_SPLIT
assertEquals(NUM_RECORDS_PER_SPLIT, getCommittedOffsetMetric(tp0, metricListener));
assertEquals(NUM_RECORDS_PER_SPLIT, getCommittedOffsetMetric(tp1, metricListener));
- // Number of successful commits should be 1
+ // Number of successful commits should be greater than 0
final Optional<Counter> commitsSucceeded =
metricListener.getCounter(
KAFKA_SOURCE_READER_METRIC_GROUP, COMMITS_SUCCEEDED_METRIC_COUNTER);
assertTrue(commitsSucceeded.isPresent());
- assertEquals(1L, commitsSucceeded.get().getCount());
+ MatcherAssert.assertThat(commitsSucceeded.get().getCount(), Matchers.greaterThan(0L));
}
}