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:40:16 UTC
[flink] branch release-1.13 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.13
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.13 by this push:
new 00bb720 [FLINK-23391][connector/kafka] Fix flaky Kafka source metric test by retrying notifyCheckpointComplete until success or timeout
00bb720 is described below
commit 00bb720f9b85880d68aa521633f17c12d4054ada
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 | 36 ++++++++++++++--------
1 file changed, 23 insertions(+), 13 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 16cab86..b88ff68 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
@@ -35,6 +35,7 @@ import org.apache.flink.connector.testutils.source.reader.SourceReaderTestBase;
import org.apache.flink.connector.testutils.source.reader.TestingReaderContext;
import org.apache.flink.connector.testutils.source.reader.TestingReaderOutput;
import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.metrics.testutils.MetricListener;
@@ -320,30 +321,39 @@ 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
- assertEquals(
- 1L,
- metricListener
- .getCounter(
- KAFKA_SOURCE_READER_METRIC_GROUP,
- COMMITS_SUCCEEDED_METRIC_COUNTER)
- .getCount());
+ // Number of successful commits should be greater than 0
+ final Counter commitsSucceeded =
+ metricListener.getCounter(
+ KAFKA_SOURCE_READER_METRIC_GROUP, COMMITS_SUCCEEDED_METRIC_COUNTER);
+ MatcherAssert.assertThat(commitsSucceeded.getCount(), Matchers.greaterThan(0L));
}
}