fapaul commented on a change in pull request #16875:
URL: https://github.com/apache/flink/pull/16875#discussion_r692798234
##########
File path:
flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterITCase.java
##########
@@ -82,49 +94,107 @@
.withLogConsumer(LOG_CONSUMER)
.withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS);
- @Before
- public void setUp() {
- metricListener = new MetricListener();
+ @BeforeAll
+ public static void beforeAll() {
+ KAFKA_CONTAINER.start();
}
- @Parameterized.Parameters
- public static List<DeliveryGuarantee> guarantees() {
- return ImmutableList.of(
- DeliveryGuarantee.NONE,
- DeliveryGuarantee.AT_LEAST_ONCE,
- DeliveryGuarantee.EXACTLY_ONCE);
+ @AfterAll
+ public static void afterAll() {
+ KAFKA_CONTAINER.stop();
}
- public KafkaWriterITCase(DeliveryGuarantee guarantee) {
- this.guarantee = guarantee;
+ @BeforeEach
+ public void setUp() {
+ metricListener = new MetricListener();
}
- @Test
- public void testRegisterMetrics() throws Exception {
+ @ParameterizedTest
+ @EnumSource(DeliveryGuarantee.class)
+ public void testRegisterMetrics(DeliveryGuarantee guarantee) throws
Exception {
try (final KafkaWriter<Integer> ignored =
- createWriterWithConfiguration(getKafkaClientConfiguration())) {
+ createWriterWithConfiguration(getKafkaClientConfiguration(),
guarantee)) {
metricListener.getGauge(KAFKA_METRIC_WITH_GROUP_NAME);
}
}
- @Test
- public void testNotRegisterMetrics() throws Exception {
+ @ParameterizedTest
+ @EnumSource(DeliveryGuarantee.class)
+ public void testNotRegisterMetrics(DeliveryGuarantee guarantee) throws
Exception {
final Properties config = getKafkaClientConfiguration();
config.put("flink.disable-metrics", "true");
- try (final KafkaWriter<Integer> ignored =
createWriterWithConfiguration(config)) {
+ try (final KafkaWriter<Integer> ignored =
+ createWriterWithConfiguration(config, guarantee)) {
assertThrows(
IllegalArgumentException.class,
() ->
metricListener.getGauge(KAFKA_METRIC_WITH_GROUP_NAME));
}
}
- private KafkaWriter<Integer> createWriterWithConfiguration(Properties
config) {
+ @Test
+ public void testIncreasingByteOutCounter() throws Exception {
+ final OperatorIOMetricGroup operatorIOMetricGroup = new
NumBytesOutOperatorIOMetricGroup();
+ final InternalSinkWriterMetricGroup metricGroup =
+ InternalSinkWriterMetricGroup.mock(
+ metricListener.getMetricGroup(),
operatorIOMetricGroup);
+ try (final KafkaWriter<Integer> writer =
+ createWriterWithConfiguration(
+ getKafkaClientConfiguration(), DeliveryGuarantee.NONE,
metricGroup)) {
+ final Counter numBytesOut =
operatorIOMetricGroup.getNumBytesOutCounter();
+ Assertions.assertEquals(numBytesOut.getCount(), 0L);
+ writer.write(1, SINK_WRITER_CONTEXT);
+ metricListener.getGauge("currentSendTime");
+ MatcherAssert.assertThat(numBytesOut.getCount(), greaterThan(0L));
+ }
+ }
+
+ @Test
+ public void testCurrentSendTimeMetric() throws Exception {
+ final OperatorIOMetricGroup operatorIOMetricGroup = new
NumBytesOutOperatorIOMetricGroup();
+ final InternalSinkWriterMetricGroup metricGroup =
+ InternalSinkWriterMetricGroup.mock(
+ metricListener.getMetricGroup(),
operatorIOMetricGroup);
+ try (final KafkaWriter<Integer> writer =
+ createWriterWithConfiguration(
+ getKafkaClientConfiguration(),
+ DeliveryGuarantee.AT_LEAST_ONCE,
+ metricGroup)) {
+ final Gauge<Long> currentSendTime =
metricListener.getGauge("currentSendTime");
+ Assertions.assertEquals(currentSendTime.getValue(), 0L);
+ IntStream.range(0, 100)
+ .forEach(
+ (run) -> {
+ try {
+ writer.write(1, SINK_WRITER_CONTEXT);
+ // Manually flush the records to generate
a sendTime
+ if (run % 10 == 0) {
+ writer.prepareCommit(false);
Review comment:
I am very hesitant to use `@VisibleForTesting` because it always feels
to me like proof of bad design if something cannot be tested without it. In
this case, I'd refrain from adding it. WDYT?
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
