tmp diag code for testing

Author: gaborgsomogyi

.github/workflows/push_pr.yml

@@ -28,12 +28,13 @@ 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_version: ${{ matrix.flink }}
       jdk_version: ${{ matrix.jdk }}
+      timeout_test: 10
   python_test:
     strategy:
       matrix:
@@ -43,3 +44,4 @@ jobs:
     with:
       flink_version: ${{ matrix.flink }}
       jdk_version: ${{ matrix.jdk }}
+      timeout_test: 10

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/sink/internal/ProducerPoolImpl.java

@@ -26,7 +26,7 @@
 import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nullable;
-import javax.annotation.concurrent.NotThreadSafe;
+import javax.annotation.concurrent.ThreadSafe;
 
 import java.util.ArrayDeque;
 import java.util.ArrayList;
@@ -68,10 +68,9 @@
  *       pool is still in the transaction or not.
  * </ul>
  *
- * <p>This pool is not thread-safe and is only intended to be accessed from the writer, which owns
- * it.
+ * <p>This pool is thread-safe and can be accessed concurrently by multiple threads.
  */
-@NotThreadSafe
+@ThreadSafe
 @Internal
 public class ProducerPoolImpl implements ProducerPool {
     private static final Logger LOG = LoggerFactory.getLogger(ProducerPoolImpl.class);
@@ -116,7 +115,7 @@ public ProducerPoolImpl(
     }
 
     @Override
-    public void recycleByTransactionId(String transactionalId, boolean success) {
+    public synchronized void recycleByTransactionId(String transactionalId, boolean success) {
         ProducerEntry producerEntry = producerByTransactionalId.remove(transactionalId);
         LOG.debug("Transaction {} finished, producer {}", transactionalId, producerEntry);
 
@@ -169,7 +168,7 @@ private void closeProducer(@Nullable FlinkKafkaInternalProducer<byte[], byte[]>
     }
 
     @Override
-    public void recycle(FlinkKafkaInternalProducer<byte[], byte[]> producer) {
+    public synchronized void recycle(FlinkKafkaInternalProducer<byte[], byte[]> producer) {
         recycleProducer(producer);
         ProducerEntry producerEntry =
                 producerByTransactionalId.remove(producer.getTransactionalId());
@@ -218,7 +217,7 @@ private void initPrecommittedTransactions(
     }
 
     @Override
-    public FlinkKafkaInternalProducer<byte[], byte[]> getTransactionalProducer(
+    public synchronized FlinkKafkaInternalProducer<byte[], byte[]> getTransactionalProducer(
             String transactionalId, long checkpointId) {
         FlinkKafkaInternalProducer<byte[], byte[]> producer = producerPool.poll();
         if (producer == null) {
@@ -249,17 +248,17 @@ public FlinkKafkaInternalProducer<byte[], byte[]> getTransactionalProducer(
     }
 
     @Override
-    public Collection<CheckpointTransaction> getOngoingTransactions() {
+    public synchronized Collection<CheckpointTransaction> getOngoingTransactions() {
         return new ArrayList<>(transactionalIdsByCheckpoint.keySet());
     }
 
     @VisibleForTesting
-    public Collection<FlinkKafkaInternalProducer<byte[], byte[]>> getProducers() {
-        return producerPool;
+    public synchronized Collection<FlinkKafkaInternalProducer<byte[], byte[]>> getProducers() {
+        return new ArrayList<>(producerPool);
     }
 
     @Override
-    public void close() throws Exception {
+    public synchronized void close() throws Exception {
         LOG.debug(
                 "Closing used producers {} and free producers {}",
                 producerByTransactionalId,

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(