From 2931c05004933be6b68bd6c5de6575670258dd41 Mon Sep 17 00:00:00 2001
From: Aleksandr Pilipenko <z3d1kk@gmail.com>
Date: Fri, 2 Feb 2024 01:16:52 +0000
Subject: [PATCH] flaky test check

---
 .github/workflows/common.yml                  | 17 ++++++++-------
 .../flink-connector-kinesis/pom.xml           |  3 +++
 .../FakeKinesisFanOutBehavioursFactory.java   | 21 ++++++++++++++++++-
 .../src/test/resources/log4j2-test.properties |  2 +-
 4 files changed, 33 insertions(+), 10 deletions(-)

diff --git a/.github/workflows/common.yml b/.github/workflows/common.yml
index c0c0bb70..cb51eb35 100644
--- a/.github/workflows/common.yml
+++ b/.github/workflows/common.yml
@@ -40,7 +40,7 @@ on:
         description: "The timeout in minutes for the compile and test step."
         required: false
         type: number
-        default: 50
+        default: 30
 
 jobs:
   compile_and_test:
@@ -97,7 +97,8 @@ jobs:
         run: |
           set -o pipefail
 
-          mvn clean install -Dflink.convergence.phase=install -Pcheck-convergence -U -B ${{ env.MVN_CONNECTION_OPTIONS }} \
+          mvn clean install -Dflink.convergence.phase=install -Pcheck-convergence \
+            -X -U -B ${{ env.MVN_CONNECTION_OPTIONS }} \
             -DaltDeploymentRepository=validation_repository::default::file:${{ env.MVN_VALIDATION_DIR }} \
             -Dflink.version=${{ inputs.flink_version }} | tee ${{ env.MVN_BUILD_OUTPUT_FILE }}
 
@@ -130,9 +131,9 @@ jobs:
 
           mvn clean
 
-      - name: Check licensing
-        run: |
-          mvn ${MVN_COMMON_OPTIONS} exec:java@check-license -N \
-            -Dexec.args="${{ env.MVN_BUILD_OUTPUT_FILE }} $(pwd) $(pwd)" \
-            ${{ env.MVN_CONNECTION_OPTIONS }} \
-            -Dlog4j.configurationFile=file://$(pwd)/tools/ci/log4j.properties
+#      - name: Check licensing
+#        run: |
+#          mvn ${MVN_COMMON_OPTIONS} exec:java@check-license -N \
+#            -Dexec.args="${{ env.MVN_BUILD_OUTPUT_FILE }} $(pwd) $(pwd)" \
+#            ${{ env.MVN_CONNECTION_OPTIONS }} \
+#            -Dlog4j.configurationFile=file://$(pwd)/tools/ci/log4j.properties
diff --git a/flink-connector-aws/flink-connector-kinesis/pom.xml b/flink-connector-aws/flink-connector-kinesis/pom.xml
index 2ffd448a..4fa54647 100644
--- a/flink-connector-aws/flink-connector-kinesis/pom.xml
+++ b/flink-connector-aws/flink-connector-kinesis/pom.xml
@@ -36,6 +36,9 @@ under the License.
         <aws.kinesis-kpl.version>0.14.1</aws.kinesis-kpl.version>
         <aws.dynamodbstreams-kinesis-adapter.version>1.5.3</aws.dynamodbstreams-kinesis-adapter.version>
         <hamcrest.version>1.3</hamcrest.version>
+
+<!--        <flink.forkCountUnitTest>1</flink.forkCountUnitTest>-->
+<!--        <flink.forkCountITCase>1</flink.forkCountITCase>-->
     </properties>
 
     <packaging>jar</packaging>
diff --git a/flink-connector-aws/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/testutils/FakeKinesisFanOutBehavioursFactory.java b/flink-connector-aws/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/testutils/FakeKinesisFanOutBehavioursFactory.java
index 9985fbf5..58de4e77 100644
--- a/flink-connector-aws/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/testutils/FakeKinesisFanOutBehavioursFactory.java
+++ b/flink-connector-aws/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/testutils/FakeKinesisFanOutBehavioursFactory.java
@@ -23,6 +23,8 @@
 import com.amazonaws.kinesis.agg.RecordAggregator;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.core.SdkBytes;
 import software.amazon.awssdk.services.kinesis.model.Consumer;
 import software.amazon.awssdk.services.kinesis.model.ConsumerDescription;
@@ -433,6 +435,8 @@ public SingleShardFanOutKinesisAsyncV2 build() {
      */
     public abstract static class AbstractSingleShardFanOutKinesisAsyncV2
             extends KinesisProxyAsyncV2InterfaceAdapter {
+        private final Logger logger =
+                LoggerFactory.getLogger(AbstractSingleShardFanOutKinesisAsyncV2.class);
 
         private final List<SubscribeToShardRequest> requests = new ArrayList<>();
         private int remainingSubscriptions;
@@ -461,16 +465,31 @@ public CompletableFuture<Void> subscribeToShard(
 
             return CompletableFuture.supplyAsync(
                     () -> {
+                        logger.info(
+                                "{}: Sending response received message",
+                                Thread.currentThread().getName());
                         responseHandler.responseReceived(
                                 SubscribeToShardResponse.builder().build());
+                        logger.info("{}: Handling event stream", Thread.currentThread().getName());
                         responseHandler.onEventStream(
                                 subscriber -> {
                                     final List<SubscribeToShardEvent> eventsToSend;
-
+                                    logger.info(
+                                            "{}: SubscribeToShardEvent, remainingSubscriptions={}",
+                                            Thread.currentThread().getName(),
+                                            remainingSubscriptions);
                                     if (remainingSubscriptions > 0) {
+                                        logger.info(
+                                                "{}: Subscription event, remainingSubscriptions={}",
+                                                Thread.currentThread().getName(),
+                                                remainingSubscriptions);
                                         eventsToSend = getEventsToSend();
                                         remainingSubscriptions--;
                                     } else {
+                                        logger.info(
+                                                "{}: end subscription event, remainingSubscriptions={}",
+                                                Thread.currentThread().getName(),
+                                                remainingSubscriptions);
                                         eventsToSend =
                                                 Collections.singletonList(
                                                         SubscribeToShardEvent.builder()
diff --git a/flink-connector-aws/flink-connector-kinesis/src/test/resources/log4j2-test.properties b/flink-connector-aws/flink-connector-kinesis/src/test/resources/log4j2-test.properties
index 6faab06b..ace06c6d 100644
--- a/flink-connector-aws/flink-connector-kinesis/src/test/resources/log4j2-test.properties
+++ b/flink-connector-aws/flink-connector-kinesis/src/test/resources/log4j2-test.properties
@@ -18,7 +18,7 @@
 
 # Set root logger level to OFF to not flood build logs
 # set manually to INFO for debugging purposes
-rootLogger.level = OFF
+rootLogger.level = INFO
 rootLogger.appenderRef.test.ref = TestLogger
 
 appender.testlogger.name = TestLogger