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 2024/10/09 09:54:40 UTC
(flink-connector-kafka) 06/06: [FLINK-36441] Fix KafkaWriterITCase#testIncreasingRecordBasedCounters
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
commit 5eeafd6a8cd99490dcbc9425c2bdb0b24dc279ec
Author: Arvid Heise <AH...@users.noreply.github.com>
AuthorDate: Fri Oct 4 10:53:28 2024 +0200
[FLINK-36441] Fix KafkaWriterITCase#testIncreasingRecordBasedCounters
The test tried to assert on byte counts which are written async. Commit adds flushing and establishes a baseline so that metadata request don't interfere with assertions.
---
.../connector/kafka/sink/KafkaWriterITCase.java | 27 ++++++++++++++--------
1 file changed, 18 insertions(+), 9 deletions(-)
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java
index e363d837..a559e3a5 100644
--- a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java
@@ -88,31 +88,40 @@ public class KafkaWriterITCase extends KafkaWriterTestBase {
try (final KafkaWriter<Integer> writer =
createWriterWithConfiguration(
- getKafkaClientConfiguration(), DeliveryGuarantee.NONE, metricGroup)) {
+ getKafkaClientConfiguration(),
+ DeliveryGuarantee.AT_LEAST_ONCE,
+ metricGroup)) {
final Counter numBytesOut = metricGroup.getIOMetricGroup().getNumBytesOutCounter();
final Counter numRecordsOut = metricGroup.getIOMetricGroup().getNumRecordsOutCounter();
final Counter numRecordsOutErrors = metricGroup.getNumRecordsOutErrorsCounter();
final Counter numRecordsSendErrors = metricGroup.getNumRecordsSendErrorsCounter();
- assertThat(numBytesOut.getCount()).isEqualTo(0L);
- assertThat(numRecordsOut.getCount()).isEqualTo(0);
+
+ // ApiVersionsRequest etc. is sent on initialization, so establish some baseline
+ writer.write(0, SINK_WRITER_CONTEXT);
+ writer.flush(false); // ensure data is actually written
+ timeService.trigger(); // sync byte count
+ long baselineCount = numBytesOut.getCount();
+ assertThat(numRecordsOut.getCount()).isEqualTo(1);
assertThat(numRecordsOutErrors.getCount()).isEqualTo(0);
assertThat(numRecordsSendErrors.getCount()).isEqualTo(0);
// elements for which the serializer returns null should be silently skipped
writer.write(null, SINK_WRITER_CONTEXT);
- timeService.trigger();
- assertThat(numBytesOut.getCount()).isEqualTo(0L);
- assertThat(numRecordsOut.getCount()).isEqualTo(0);
+ writer.flush(false);
+ timeService.trigger(); // sync byte count
+ assertThat(numBytesOut.getCount()).isEqualTo(baselineCount);
+ assertThat(numRecordsOut.getCount()).isEqualTo(1);
assertThat(numRecordsOutErrors.getCount()).isEqualTo(0);
assertThat(numRecordsSendErrors.getCount()).isEqualTo(0);
// but elements for which a non-null producer record is returned should count
writer.write(1, SINK_WRITER_CONTEXT);
- timeService.trigger();
- assertThat(numRecordsOut.getCount()).isEqualTo(1);
+ writer.flush(false);
+ timeService.trigger(); // sync byte count
+ assertThat(numBytesOut.getCount()).isGreaterThan(baselineCount);
+ assertThat(numRecordsOut.getCount()).isEqualTo(2);
assertThat(numRecordsOutErrors.getCount()).isEqualTo(0);
assertThat(numRecordsSendErrors.getCount()).isEqualTo(0);
- assertThat(numBytesOut.getCount()).isGreaterThan(0L);
}
}