tmp diag code for testing

Author: gaborgsomogyi

.github/workflows/push_pr.yml

@@ -28,7 +28,7 @@ jobs:
   compile_and_test:
     strategy:
       matrix:
-        flink: &flink_versions [ 2.0.1, 2.1.1, 2.2.0 ]
+        flink: &flink_versions [ 2.2.0 ]
         jdk: &jdk_versions [ '11, 17, 21' ]
     uses: apache/flink-connector-shared-utils/.github/workflows/ci.yml@ci_utils
     with:

flink-connector-kafka/pom.xml

@@ -421,6 +421,15 @@ under the License.
 					</execution>
 				</executions>
 			</plugin>
+
+			<!-- Temporarily run only KafkaSinkITCase for faster feedback on hanging test issue -->
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-surefire-plugin</artifactId>
+				<configuration>
+					<test>KafkaSinkITCase</test>
+				</configuration>
+			</plugin>
 		</plugins>
 	</build>
 

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/util/AdminUtils.java

@@ -72,7 +72,17 @@ public static Set<String> getTopicsByPattern(Admin admin, Pattern topicPattern)
     public static Map<String, TopicDescription> getTopicMetadata(
             Admin admin, Collection<String> topicNames) {
         try {
-            return admin.describeTopics(topicNames).allTopicNames().get();
+            // Add timeout to prevent infinite hang during recovery with POOLING strategy +
+            // chained=false
+            return admin.describeTopics(topicNames)
+                    .allTopicNames()
+                    .get(30, java.util.concurrent.TimeUnit.SECONDS);
+        } catch (java.util.concurrent.TimeoutException e) {
+            throw new RuntimeException(
+                    String.format(
+                            "Timeout while getting metadata for topics %s. This may indicate network issues or slow Kafka responses during recovery.",
+                            topicNames),
+                    e);
         } catch (Exception e) {
             checkIfInterrupted(e);
             throw new RuntimeException(
@@ -83,7 +93,17 @@ public static Map<String, TopicDescription> getTopicMetadata(
     public static Map<TopicPartition, DescribeProducersResult.PartitionProducerState>
             getProducerStates(Admin admin, Collection<String> topicNames) {
         try {
-            return admin.describeProducers(getTopicPartitions(admin, topicNames)).all().get();
+            // Add timeout to prevent infinite hang during recovery with POOLING strategy +
+            // chained=false
+            return admin.describeProducers(getTopicPartitions(admin, topicNames))
+                    .all()
+                    .get(30, java.util.concurrent.TimeUnit.SECONDS);
+        } catch (java.util.concurrent.TimeoutException e) {
+            throw new RuntimeException(
+                    String.format(
+                            "Timeout while getting producers for topics %s. This may indicate network issues or slow Kafka responses during recovery.",
+                            topicNames),
+                    e);
         } catch (Exception e) {
             checkIfInterrupted(e);
             throw new RuntimeException(
@@ -103,12 +123,20 @@ public static Collection<Long> getProducerIds(Admin admin, Collection<String> to
     public static Collection<TransactionListing> getOpenTransactionsForTopics(
             Admin admin, Collection<String> topicNames) {
         try {
+            // Add timeout to prevent infinite hang during recovery with POOLING strategy +
+            // chained=false
             return admin.listTransactions(
                             new ListTransactionsOptions()
                                     .filterProducerIds(getProducerIds(admin, topicNames))
                                     .filterStates(List.of(TransactionState.ONGOING)))
                     .all()
-                    .get();
+                    .get(30, java.util.concurrent.TimeUnit.SECONDS);
+        } catch (java.util.concurrent.TimeoutException e) {
+            throw new RuntimeException(
+                    String.format(
+                            "Timeout while getting open transactions for topics %s. This may indicate network issues or slow Kafka responses during recovery.",
+                            topicNames),
+                    e);
         } catch (Exception e) {
             checkIfInterrupted(e);
             throw new RuntimeException(

flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java

@@ -282,13 +282,21 @@ public void testRecoveryWithAtLeastOnceGuarantee() throws Exception {
     @MethodSource("getEOSParameters")
     public void testRecoveryWithExactlyOnceGuarantee(
             TransactionNamingStrategy namingStrategy, boolean chained) throws Exception {
+        LOG.info(
+                "========== STARTING testRecoveryWithExactlyOnceGuarantee with namingStrategy={}, chained={} ==========",
+                namingStrategy,
+                chained);
         testRecoveryWithAssertion(DeliveryGuarantee.EXACTLY_ONCE, 1, namingStrategy, chained);
     }
 
     @ParameterizedTest(name = "{0}, chained={1}")
     @MethodSource("getEOSParameters")
     public void testRecoveryWithExactlyOnceGuaranteeAndConcurrentCheckpoints(
             TransactionNamingStrategy namingStrategy, boolean chained) throws Exception {
+        LOG.info(
+                "========== STARTING testRecoveryWithExactlyOnceGuaranteeAndConcurrentCheckpoints with namingStrategy={}, chained={} ==========",
+                namingStrategy,
+                chained);
         testRecoveryWithAssertion(DeliveryGuarantee.EXACTLY_ONCE, 2, namingStrategy, chained);
     }
 
@@ -653,7 +661,15 @@ private void testRecoveryWithAssertion(
                         .setTransactionalIdPrefix("kafka-sink")
                         .setTransactionNamingStrategy(namingStrategy)
                         .build());
+        LOG.info(
+                "========== ABOUT TO CALL env.execute() with namingStrategy={}, chained={} ==========",
+                namingStrategy,
+                chained);
         env.execute();
+        LOG.info(
+                "========== SUCCESSFULLY COMPLETED env.execute() with namingStrategy={}, chained={} ==========",
+                namingStrategy,
+                chained);
 
         List<Long> committedRecords =
                 deserializeValues(

pom.xml

@@ -580,6 +580,9 @@ under the License.
 			<plugin>
 				<groupId>com.diffplug.spotless</groupId>
 				<artifactId>spotless-maven-plugin</artifactId>
+				<configuration>
+					<skip>false</skip>
+				</configuration>
 			</plugin>
 			<plugin>
 				<groupId>org.apache.maven.plugins</groupId>