Merge branch 'main' into feature/kafka4

Author: Shekharrajak

.gitignore

@@ -6,7 +6,6 @@ scalastyle-output.xml
 .idea/*
 !.idea/vcs.xml
 .vscode
-
 .metadata
 .settings
 .project

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/FlinkKafkaInternalProducer.java

@@ -179,7 +179,7 @@ public long getProducerId() {
     public void setTransactionId(String transactionalId) {
         checkState(
                 !isInTransaction(),
-                String.format("Another transaction %s is still open.", transactionalId));
+                String.format("Another transaction %s is still open.", this.transactionalId));
         LOG.debug("Change transaction id from {} to {}", this.transactionalId, transactionalId);
         this.transactionalId = transactionalId;
         Object transactionManager = getTransactionManager();

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumerator.java

@@ -32,7 +32,6 @@
 
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.KafkaAdminClient;
-
 import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsSpec;
 import org.apache.kafka.clients.admin.ListOffsetsResult;
 import org.apache.kafka.clients.admin.OffsetSpec;

flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/testutils/KafkaUtil.java

@@ -67,7 +67,6 @@ public static KafkaContainer createKafkaContainer(String containerName) {
                 .withEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1")
                 .withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1")
                 .withEnv("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1")
-                .withEnv("KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE", "false")
                 .withEnv("KAFKA_LOG4J_ROOT_LOGLEVEL", logLevel)
                 .withEnv("KAFKA_LOG4J_LOGGERS", "state.change.logger=" + logLevel)
                 .withEnv("KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE", "false")