[kafka] branch 3.4 updated: KAFKA-14938: Fixing flaky test testConnectorBoundary (#13646)

[ce...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

cegerton pushed a commit to branch 3.4
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/3.4 by this push:
     new 3808bda6794 KAFKA-14938: Fixing flaky test testConnectorBoundary (#13646)
3808bda6794 is described below

commit 3808bda67940c90764b7645482d333841ed33e23
Author: Sambhav Jain <67...@users.noreply.github.com>
AuthorDate: Wed Jul 12 21:15:49 2023 +0530

    KAFKA-14938: Fixing flaky test testConnectorBoundary (#13646)
    
    Reviewers: Sagar Rao <sa...@gmail.com>, Yash Mayya <ya...@gmail.com>,
    Sudesh Wasnik <sw...@confluent.io>, Chris Egerton <ch...@aiven.io>
---
 .../integration/ExactlyOnceSourceIntegrationTest.java   | 17 ++++++++++-------
 .../connect/integration/MonitorableSourceConnector.java |  4 ++++
 2 files changed, 14 insertions(+), 7 deletions(-)

diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExactlyOnceSourceIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExactlyOnceSourceIntegrationTest.java
index a2dd730b344..b717391745c 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExactlyOnceSourceIntegrationTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExactlyOnceSourceIntegrationTest.java
@@ -388,9 +388,12 @@ public class ExactlyOnceSourceIntegrationTest {
         props.put(CUSTOM_TRANSACTION_BOUNDARIES_CONFIG, MonitorableSourceConnector.TRANSACTION_BOUNDARIES_SUPPORTED);
         props.put(MESSAGES_PER_POLL_CONFIG, Integer.toString(recordsProduced));
 
-        // expect all records to be consumed and committed by the connector
-        connectorHandle.expectedRecords(recordsProduced);
-        connectorHandle.expectedCommits(recordsProduced);
+        // the connector aborts some transactions, which causes records that it has emitted (and for which
+        // SourceTask::commitRecord has been invoked) to be invisible to consumers; we expect the task to
+        // emit at most 233 records in total before 100 records have been emitted as part of one or more
+        // committed transactions
+        connectorHandle.expectedRecords(233);
+        connectorHandle.expectedCommits(233);
 
         // start a source connector
         connect.configureConnector(CONNECTOR_NAME, props);
@@ -407,10 +410,10 @@ public class ExactlyOnceSourceIntegrationTest {
         consumerProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
         // consume all records from the source topic or fail, to ensure that they were correctly produced
         ConsumerRecords<byte[], byte[]> sourceRecords = connect.kafka().consumeAll(
-                CONSUME_RECORDS_TIMEOUT_MS,
-                Collections.singletonMap(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed"),
-                null,
-                topic
+            CONSUME_RECORDS_TIMEOUT_MS,
+            Collections.singletonMap(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed"),
+            null,
+            topic
         );
         assertTrue("Not enough records produced by source connector. Expected at least: " + recordsProduced + " + but got " + sourceRecords.count(),
                 sourceRecords.count() >= recordsProduced);
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSourceConnector.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSourceConnector.java
index 33ba1588a7d..83df2133279 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSourceConnector.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSourceConnector.java
@@ -255,13 +255,17 @@ public class MonitorableSourceConnector extends SampleSourceConnector {
             if (context.transactionContext() == null || seqno != nextTransactionBoundary) {
                 return;
             }
+            long transactionSize = nextTransactionBoundary - priorTransactionBoundary;
+
             // If the transaction boundary ends on an even-numbered offset, abort it
             // Otherwise, commit
             boolean abort = nextTransactionBoundary % 2 == 0;
             calculateNextBoundary();
             if (abort) {
+                log.info("Aborting transaction of {} records", transactionSize);
                 context.transactionContext().abortTransaction(record);
             } else {
+                log.info("Committing transaction of {} records", transactionSize);
                 context.transactionContext().commitTransaction(record);
             }
         }