diff --git a/flink-connector-kafka/archunit-violations/86dfd459-67a9-4b26-9b5c-0b0bbf22681a b/flink-connector-kafka/archunit-violations/86dfd459-67a9-4b26-9b5c-0b0bbf22681a
index 07efd19a9..2544a7521 100644
--- a/flink-connector-kafka/archunit-violations/86dfd459-67a9-4b26-9b5c-0b0bbf22681a
+++ b/flink-connector-kafka/archunit-violations/86dfd459-67a9-4b26-9b5c-0b0bbf22681a
@@ -1,101 +1,5 @@
 org.apache.flink.connector.kafka.sink.FlinkKafkaInternalProducerITCase does not satisfy: only one of the following predicates match:\
 * reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension\
-* reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\
- or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule
-org.apache.flink.connector.kafka.sink.KafkaSinkITCase does not satisfy: only one of the following predicates match:\
-* reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension\
-* reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\
- or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule
-org.apache.flink.connector.kafka.sink.KafkaTransactionLogITCase does not satisfy: only one of the following predicates match:\
-* reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension\
-* reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\
- or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule
-org.apache.flink.connector.kafka.sink.KafkaWriterITCase does not satisfy: only one of the following predicates match:\
-* reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension\
-* reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\
- or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule
-org.apache.flink.connector.kafka.sink.KafkaWriterITCase does not satisfy: only one of the following predicates match:\
-* reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension or are , and of type MiniClusterTestEnvironment and annotated with @TestEnv\
-* reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\
- or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule
-org.apache.flink.connector.kafka.sink.KafkaWriterFaultToleranceITCase does not satisfy: only one of the following predicates match:\
-* reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension\
-* reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\
- or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule
-org.apache.flink.connector.kafka.sink.KafkaWriterITCase does not satisfy: only one of the following predicates match:
-* reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension
-* reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension or are , and of type MiniClusterTestEnvironment and annotated with @TestEnv
-* reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension
-* reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension
- or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule
-org.apache.flink.connector.kafka.source.KafkaSourceITCase does not satisfy: only one of the following predicates match:\
-* reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension\
-* reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\
- or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule
-org.apache.flink.connector.kafka.source.KafkaSourceLegacyITCase does not satisfy: only one of the following predicates match:\
-* reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension\
-* reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\
- or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule
-org.apache.flink.streaming.connectors.kafka.FlinkKafkaInternalProducerITCase does not satisfy: only one of the following predicates match:\
-* reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension\
-* reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\
- or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule
-org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducerITCase does not satisfy: only one of the following predicates match:\
-* reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension\
-* reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\
- or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule
-org.apache.flink.streaming.connectors.kafka.KafkaITCase does not satisfy: only one of the following predicates match:\
-* reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension\
-* reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\
- or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule
-org.apache.flink.streaming.connectors.kafka.KafkaProducerAtLeastOnceITCase does not satisfy: only one of the following predicates match:\
-* reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension\
-* reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\
- or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule
-org.apache.flink.streaming.connectors.kafka.KafkaProducerExactlyOnceITCase does not satisfy: only one of the following predicates match:\
-* reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension\
-* reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\
- or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule
-org.apache.flink.streaming.connectors.kafka.shuffle.KafkaShuffleExactlyOnceITCase does not satisfy: only one of the following predicates match:\
-* reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension\
-* reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\
- or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule
-org.apache.flink.streaming.connectors.kafka.shuffle.KafkaShuffleITCase does not satisfy: only one of the following predicates match:\
-* reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension\
-* reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\
-* reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\
- or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule
-org.apache.flink.connector.kafka.sink.FlinkKafkaInternalProducerITCase does not satisfy: only one of the following predicates match:\
-* reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\
 * reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension or are , and of type MiniClusterTestEnvironment and annotated with @TestEnv\
 * reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\
 * reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\
@@ -112,14 +16,13 @@ org.apache.flink.connector.kafka.sink.KafkaTransactionLogITCase does not satisfy
 * reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\
 * reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\
  or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule
-org.apache.flink.connector.kafka.sink.KafkaWriterITCase does not satisfy: only one of the following predicates match:\
 org.apache.flink.connector.kafka.sink.KafkaWriterFaultToleranceITCase does not satisfy: only one of the following predicates match:\
 * reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\
 * reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension or are , and of type MiniClusterTestEnvironment and annotated with @TestEnv\
 * reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\
 * reside outside of package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class MiniClusterExtension\
  or contain any fields that are public, static, and of type MiniClusterWithClientResource and final and annotated with @ClassRule or contain any fields that is of type MiniClusterWithClientResource and public and final and not static and annotated with @Rule
-org.apache.flink.connector.kafka.sink.KafkaWriterFaultToleranceITCase does not satisfy: only one of the following predicates match:\
+org.apache.flink.connector.kafka.sink.KafkaWriterITCase does not satisfy: only one of the following predicates match:\
 * reside in a package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type InternalMiniClusterExtension and annotated with @RegisterExtension\
 * reside outside of package 'org.apache.flink.runtime.*' and contain any fields that are static, final, and of type MiniClusterExtension and annotated with @RegisterExtension or are , and of type MiniClusterTestEnvironment and annotated with @TestEnv\
 * reside in a package 'org.apache.flink.runtime.*' and is annotated with @ExtendWith with class InternalMiniClusterExtension\
diff --git a/flink-connector-kafka/archunit-violations/c0d94764-76a0-4c50-b617-70b1754c4612 b/flink-connector-kafka/archunit-violations/c0d94764-76a0-4c50-b617-70b1754c4612
index 20326f5ec..756b92768 100644
--- a/flink-connector-kafka/archunit-violations/c0d94764-76a0-4c50-b617-70b1754c4612
+++ b/flink-connector-kafka/archunit-violations/c0d94764-76a0-4c50-b617-70b1754c4612
@@ -24,6 +24,7 @@ Constructor <org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer$Tran
 Constructor <org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher$PeriodicWatermarkEmitter.<init>(java.lang.Object, java.util.List, org.apache.flink.api.common.eventtime.WatermarkOutputMultiplexer, org.apache.flink.streaming.runtime.tasks.ProcessingTimeService, long)> has parameter of type <org.apache.flink.api.common.eventtime.WatermarkOutputMultiplexer> in (AbstractFetcher.java:0)
 Constructor <org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher$PeriodicWatermarkEmitter.<init>(java.lang.Object, java.util.List, org.apache.flink.api.common.eventtime.WatermarkOutputMultiplexer, org.apache.flink.streaming.runtime.tasks.ProcessingTimeService, long)> has parameter of type <org.apache.flink.streaming.runtime.tasks.ProcessingTimeService> in (AbstractFetcher.java:0)
 Constructor <org.apache.flink.streaming.connectors.kafka.internals.KafkaShuffleFetcher$KafkaShuffleElementDeserializer.<init>(org.apache.flink.api.common.typeutils.TypeSerializer)> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (KafkaShuffleFetcher.java:0)
+Constructor <org.apache.flink.streaming.connectors.kafka.table.DynamicKafkaRecordSerializationSchema.<init>(java.util.List, java.util.regex.Pattern, org.apache.flink.connector.kafka.sink.KafkaPartitioner, org.apache.flink.api.common.serialization.SerializationSchema, org.apache.flink.api.common.serialization.SerializationSchema, [Lorg.apache.flink.table.data.RowData$FieldGetter;, [Lorg.apache.flink.table.data.RowData$FieldGetter;, boolean, [I, boolean)> has parameter of type <[Lorg.apache.flink.table.data.RowData$FieldGetter;> in (DynamicKafkaRecordSerializationSchema.java:0)
 Field <org.apache.flink.connector.kafka.dynamic.source.metrics.KafkaClusterMetricGroupManager.metricGroups> has generic type <java.util.Map<java.lang.String, org.apache.flink.runtime.metrics.groups.AbstractMetricGroup>> with type argument depending on <org.apache.flink.runtime.metrics.groups.AbstractMetricGroup> in (KafkaClusterMetricGroupManager.java:0)
 Field <org.apache.flink.connector.kafka.dynamic.source.reader.DynamicKafkaSourceReader.availabilityHelper> has type <org.apache.flink.streaming.runtime.io.MultipleFuturesAvailabilityHelper> in (DynamicKafkaSourceReader.java:0)
 Field <org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher$PeriodicWatermarkEmitter.timerService> has type <org.apache.flink.streaming.runtime.tasks.ProcessingTimeService> in (AbstractFetcher.java:0)
@@ -99,112 +100,3 @@ Method <org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptionsU
 Method <org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSink.createSerialization(org.apache.flink.table.connector.sink.DynamicTableSink$Context, org.apache.flink.table.connector.format.EncodingFormat, [I, java.lang.String)> calls method <org.apache.flink.table.types.utils.DataTypeUtils.stripRowPrefix(org.apache.flink.table.types.DataType, java.lang.String)> in (KafkaDynamicSink.java:386)
 Method <org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSink.getFieldGetters(java.util.List, [I)> has return type <[Lorg.apache.flink.table.data.RowData$FieldGetter;> in (KafkaDynamicSink.java:0)
 Method <org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSource.createDeserialization(org.apache.flink.table.connector.source.DynamicTableSource$Context, org.apache.flink.table.connector.format.DecodingFormat, [I, java.lang.String)> calls method <org.apache.flink.table.types.utils.DataTypeUtils.stripRowPrefix(org.apache.flink.table.types.DataType, java.lang.String)> in (KafkaDynamicSource.java:566)
-Constructor <org.apache.flink.connector.kafka.dynamic.source.enumerator.StoppableKafkaEnumContextProxy.<init>(java.lang.String, org.apache.flink.connector.kafka.dynamic.metadata.KafkaMetadataService, org.apache.flink.api.connector.source.SplitEnumeratorContext, java.lang.Runnable)> calls constructor <org.apache.flink.util.concurrent.ExecutorThreadFactory.<init>(java.lang.String)> in (StoppableKafkaEnumContextProxy.java:95)
-Constructor <org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchemaBuilder$KafkaRecordSerializationSchemaWrapper.<init>(java.util.function.Function, org.apache.flink.api.common.serialization.SerializationSchema, org.apache.flink.api.common.serialization.SerializationSchema, org.apache.flink.connector.kafka.sink.KafkaPartitioner, org.apache.flink.connector.kafka.sink.HeaderProvider)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (KafkaRecordSerializationSchemaBuilder.java:322)
-Constructor <org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchemaBuilder$KafkaRecordSerializationSchemaWrapper.<init>(java.util.function.Function, org.apache.flink.api.common.serialization.SerializationSchema, org.apache.flink.api.common.serialization.SerializationSchema, org.apache.flink.connector.kafka.sink.KafkaPartitioner, org.apache.flink.connector.kafka.sink.HeaderProvider)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (KafkaRecordSerializationSchemaBuilder.java:323)
-Constructor <org.apache.flink.connector.kafka.sink.KafkaSerializerWrapper.<init>(java.lang.Class, boolean, java.util.Map, java.util.function.Function)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (KafkaSerializerWrapper.java:51)
-Constructor <org.apache.flink.connector.kafka.sink.KafkaSerializerWrapper.<init>(java.lang.Class, boolean, java.util.Map, java.util.function.Function)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (KafkaSerializerWrapper.java:53)
-Constructor <org.apache.flink.connector.kafka.sink.KafkaSerializerWrapper.<init>(java.lang.Class, boolean, java.util.Map, java.util.function.Function)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (KafkaSerializerWrapper.java:54)
-Constructor <org.apache.flink.connector.kafka.sink.KafkaWriter.<init>(org.apache.flink.connector.base.DeliveryGuarantee, java.util.Properties, java.lang.String, org.apache.flink.api.connector.sink2.Sink$InitContext, org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema, org.apache.flink.api.common.serialization.SerializationSchema$InitializationContext, java.util.Collection)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaWriter.java:134)
-Constructor <org.apache.flink.connector.kafka.sink.KafkaWriter.<init>(org.apache.flink.connector.base.DeliveryGuarantee, java.util.Properties, java.lang.String, org.apache.flink.api.connector.sink2.Sink$InitContext, org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema, org.apache.flink.api.common.serialization.SerializationSchema$InitializationContext, java.util.Collection)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaWriter.java:135)
-Constructor <org.apache.flink.connector.kafka.sink.KafkaWriter.<init>(org.apache.flink.connector.base.DeliveryGuarantee, java.util.Properties, java.lang.String, org.apache.flink.api.connector.sink2.Sink$InitContext, org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema, org.apache.flink.api.common.serialization.SerializationSchema$InitializationContext, java.util.Collection)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaWriter.java:136)
-Constructor <org.apache.flink.connector.kafka.sink.KafkaWriter.<init>(org.apache.flink.connector.base.DeliveryGuarantee, java.util.Properties, java.lang.String, org.apache.flink.api.connector.sink2.Sink$InitContext, org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema, org.apache.flink.api.common.serialization.SerializationSchema$InitializationContext, java.util.Collection)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaWriter.java:137)
-Constructor <org.apache.flink.connector.kafka.sink.KafkaWriter.<init>(org.apache.flink.connector.base.DeliveryGuarantee, java.util.Properties, java.lang.String, org.apache.flink.api.connector.sink2.Sink$InitContext, org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema, org.apache.flink.api.common.serialization.SerializationSchema$InitializationContext, java.util.Collection)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaWriter.java:138)
-Constructor <org.apache.flink.connector.kafka.sink.KafkaWriter.<init>(org.apache.flink.connector.base.DeliveryGuarantee, java.util.Properties, java.lang.String, org.apache.flink.api.connector.sink2.Sink$InitContext, org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema, org.apache.flink.api.common.serialization.SerializationSchema$InitializationContext, java.util.Collection)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaWriter.java:173)
-Constructor <org.apache.flink.connector.kafka.sink.KafkaWriterState.<init>(java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaWriterState.java:28)
-Constructor <org.apache.flink.connector.kafka.sink.Recyclable.<init>(java.lang.Object, java.util.function.Consumer)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (Recyclable.java:31)
-Constructor <org.apache.flink.connector.kafka.sink.Recyclable.<init>(java.lang.Object, java.util.function.Consumer)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (Recyclable.java:32)
-Constructor <org.apache.flink.connector.kafka.sink.TransactionAborter.<init>(int, int, java.util.function.Function, java.util.function.Consumer)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (TransactionAborter.java:60)
-Constructor <org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer$KafkaTransactionContext.<init>(java.util.Set)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (FlinkKafkaProducer.java:1591)
-Constructor <org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher$PeriodicWatermarkEmitter.<init>(java.lang.Object, java.util.List, org.apache.flink.api.common.eventtime.WatermarkOutputMultiplexer, org.apache.flink.streaming.runtime.tasks.ProcessingTimeService, long)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (AbstractFetcher.java:593)
-Constructor <org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher$PeriodicWatermarkEmitter.<init>(java.lang.Object, java.util.List, org.apache.flink.api.common.eventtime.WatermarkOutputMultiplexer, org.apache.flink.streaming.runtime.tasks.ProcessingTimeService, long)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (AbstractFetcher.java:595)
-Constructor <org.apache.flink.streaming.connectors.kafka.internals.KafkaConsumerThread$CommitCallback.<init>(org.apache.flink.streaming.connectors.kafka.internals.KafkaConsumerThread, org.apache.flink.streaming.connectors.kafka.internals.KafkaCommitCallback)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (KafkaConsumerThread.java:540)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.DynamicKafkaDeserializationSchema.<init>(int, org.apache.flink.api.common.serialization.DeserializationSchema, [I, org.apache.flink.api.common.serialization.DeserializationSchema, [I, boolean, [Lorg.apache.flink.streaming.connectors.kafka.table.DynamicKafkaDeserializationSchema$MetadataConverter;, org.apache.flink.api.common.typeinfo.TypeInformation, boolean)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (DynamicKafkaDeserializationSchema.java:72)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.DynamicKafkaRecordSerializationSchema.<init>(java.util.List, java.util.regex.Pattern, org.apache.flink.connector.kafka.sink.KafkaPartitioner, org.apache.flink.api.common.serialization.SerializationSchema, org.apache.flink.api.common.serialization.SerializationSchema, [Lorg.apache.flink.table.data.RowData$FieldGetter;, [Lorg.apache.flink.table.data.RowData$FieldGetter;, boolean, [I, boolean)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (DynamicKafkaRecordSerializationSchema.java:71)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.DynamicKafkaRecordSerializationSchema.<init>(java.util.List, java.util.regex.Pattern, org.apache.flink.connector.kafka.sink.KafkaPartitioner, org.apache.flink.api.common.serialization.SerializationSchema, org.apache.flink.api.common.serialization.SerializationSchema, [Lorg.apache.flink.table.data.RowData$FieldGetter;, [Lorg.apache.flink.table.data.RowData$FieldGetter;, boolean, [I, boolean)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (DynamicKafkaRecordSerializationSchema.java:75)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.DynamicKafkaRecordSerializationSchema.<init>(java.util.List, java.util.regex.Pattern, org.apache.flink.connector.kafka.sink.KafkaPartitioner, org.apache.flink.api.common.serialization.SerializationSchema, org.apache.flink.api.common.serialization.SerializationSchema, [Lorg.apache.flink.table.data.RowData$FieldGetter;, [Lorg.apache.flink.table.data.RowData$FieldGetter;, boolean, [I, boolean)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (DynamicKafkaRecordSerializationSchema.java:87)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.DynamicKafkaRecordSerializationSchema.<init>(java.util.List, java.util.regex.Pattern, org.apache.flink.connector.kafka.sink.KafkaPartitioner, org.apache.flink.api.common.serialization.SerializationSchema, org.apache.flink.api.common.serialization.SerializationSchema, [Lorg.apache.flink.table.data.RowData$FieldGetter;, [Lorg.apache.flink.table.data.RowData$FieldGetter;, boolean, [I, boolean)> has parameter of type <[Lorg.apache.flink.table.data.RowData$FieldGetter;> in (DynamicKafkaRecordSerializationSchema.java:0)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSink.<init>(org.apache.flink.table.types.DataType, org.apache.flink.table.types.DataType, org.apache.flink.table.connector.format.EncodingFormat, org.apache.flink.table.connector.format.EncodingFormat, [I, [I, java.lang.String, java.util.List, java.util.regex.Pattern, java.util.Properties, org.apache.flink.connector.kafka.sink.KafkaPartitioner, org.apache.flink.connector.base.DeliveryGuarantee, boolean, org.apache.flink.streaming.connectors.kafka.table.SinkBufferFlushMode, java.lang.Integer, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (KafkaDynamicSink.java:181)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSink.<init>(org.apache.flink.table.types.DataType, org.apache.flink.table.types.DataType, org.apache.flink.table.connector.format.EncodingFormat, org.apache.flink.table.connector.format.EncodingFormat, [I, [I, java.lang.String, java.util.List, java.util.regex.Pattern, java.util.Properties, org.apache.flink.connector.kafka.sink.KafkaPartitioner, org.apache.flink.connector.base.DeliveryGuarantee, boolean, org.apache.flink.streaming.connectors.kafka.table.SinkBufferFlushMode, java.lang.Integer, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaDynamicSink.java:161)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSink.<init>(org.apache.flink.table.types.DataType, org.apache.flink.table.types.DataType, org.apache.flink.table.connector.format.EncodingFormat, org.apache.flink.table.connector.format.EncodingFormat, [I, [I, java.lang.String, java.util.List, java.util.regex.Pattern, java.util.Properties, org.apache.flink.connector.kafka.sink.KafkaPartitioner, org.apache.flink.connector.base.DeliveryGuarantee, boolean, org.apache.flink.streaming.connectors.kafka.table.SinkBufferFlushMode, java.lang.Integer, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaDynamicSink.java:163)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSink.<init>(org.apache.flink.table.types.DataType, org.apache.flink.table.types.DataType, org.apache.flink.table.connector.format.EncodingFormat, org.apache.flink.table.connector.format.EncodingFormat, [I, [I, java.lang.String, java.util.List, java.util.regex.Pattern, java.util.Properties, org.apache.flink.connector.kafka.sink.KafkaPartitioner, org.apache.flink.connector.base.DeliveryGuarantee, boolean, org.apache.flink.streaming.connectors.kafka.table.SinkBufferFlushMode, java.lang.Integer, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaDynamicSink.java:166)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSink.<init>(org.apache.flink.table.types.DataType, org.apache.flink.table.types.DataType, org.apache.flink.table.connector.format.EncodingFormat, org.apache.flink.table.connector.format.EncodingFormat, [I, [I, java.lang.String, java.util.List, java.util.regex.Pattern, java.util.Properties, org.apache.flink.connector.kafka.sink.KafkaPartitioner, org.apache.flink.connector.base.DeliveryGuarantee, boolean, org.apache.flink.streaming.connectors.kafka.table.SinkBufferFlushMode, java.lang.Integer, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaDynamicSink.java:167)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSink.<init>(org.apache.flink.table.types.DataType, org.apache.flink.table.types.DataType, org.apache.flink.table.connector.format.EncodingFormat, org.apache.flink.table.connector.format.EncodingFormat, [I, [I, java.lang.String, java.util.List, java.util.regex.Pattern, java.util.Properties, org.apache.flink.connector.kafka.sink.KafkaPartitioner, org.apache.flink.connector.base.DeliveryGuarantee, boolean, org.apache.flink.streaming.connectors.kafka.table.SinkBufferFlushMode, java.lang.Integer, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaDynamicSink.java:168)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSink.<init>(org.apache.flink.table.types.DataType, org.apache.flink.table.types.DataType, org.apache.flink.table.connector.format.EncodingFormat, org.apache.flink.table.connector.format.EncodingFormat, [I, [I, java.lang.String, java.util.List, java.util.regex.Pattern, java.util.Properties, org.apache.flink.connector.kafka.sink.KafkaPartitioner, org.apache.flink.connector.base.DeliveryGuarantee, boolean, org.apache.flink.streaming.connectors.kafka.table.SinkBufferFlushMode, java.lang.Integer, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaDynamicSink.java:176)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSink.<init>(org.apache.flink.table.types.DataType, org.apache.flink.table.types.DataType, org.apache.flink.table.connector.format.EncodingFormat, org.apache.flink.table.connector.format.EncodingFormat, [I, [I, java.lang.String, java.util.List, java.util.regex.Pattern, java.util.Properties, org.apache.flink.connector.kafka.sink.KafkaPartitioner, org.apache.flink.connector.base.DeliveryGuarantee, boolean, org.apache.flink.streaming.connectors.kafka.table.SinkBufferFlushMode, java.lang.Integer, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaDynamicSink.java:179)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSource.<init>(org.apache.flink.table.types.DataType, org.apache.flink.table.connector.format.DecodingFormat, org.apache.flink.table.connector.format.DecodingFormat, [I, [I, java.lang.String, java.util.List, java.util.regex.Pattern, java.util.Properties, org.apache.flink.streaming.connectors.kafka.config.StartupMode, java.util.Map, long, org.apache.flink.streaming.connectors.kafka.config.BoundedMode, java.util.Map, long, boolean, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (KafkaDynamicSource.java:210)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSource.<init>(org.apache.flink.table.types.DataType, org.apache.flink.table.connector.format.DecodingFormat, org.apache.flink.table.connector.format.DecodingFormat, [I, [I, java.lang.String, java.util.List, java.util.regex.Pattern, java.util.Properties, org.apache.flink.streaming.connectors.kafka.config.StartupMode, java.util.Map, long, org.apache.flink.streaming.connectors.kafka.config.BoundedMode, java.util.Map, long, boolean, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaDynamicSource.java:194)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSource.<init>(org.apache.flink.table.types.DataType, org.apache.flink.table.connector.format.DecodingFormat, org.apache.flink.table.connector.format.DecodingFormat, [I, [I, java.lang.String, java.util.List, java.util.regex.Pattern, java.util.Properties, org.apache.flink.streaming.connectors.kafka.config.StartupMode, java.util.Map, long, org.apache.flink.streaming.connectors.kafka.config.BoundedMode, java.util.Map, long, boolean, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaDynamicSource.java:198)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSource.<init>(org.apache.flink.table.types.DataType, org.apache.flink.table.connector.format.DecodingFormat, org.apache.flink.table.connector.format.DecodingFormat, [I, [I, java.lang.String, java.util.List, java.util.regex.Pattern, java.util.Properties, org.apache.flink.streaming.connectors.kafka.config.StartupMode, java.util.Map, long, org.apache.flink.streaming.connectors.kafka.config.BoundedMode, java.util.Map, long, boolean, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaDynamicSource.java:201)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSource.<init>(org.apache.flink.table.types.DataType, org.apache.flink.table.connector.format.DecodingFormat, org.apache.flink.table.connector.format.DecodingFormat, [I, [I, java.lang.String, java.util.List, java.util.regex.Pattern, java.util.Properties, org.apache.flink.streaming.connectors.kafka.config.StartupMode, java.util.Map, long, org.apache.flink.streaming.connectors.kafka.config.BoundedMode, java.util.Map, long, boolean, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaDynamicSource.java:203)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSource.<init>(org.apache.flink.table.types.DataType, org.apache.flink.table.connector.format.DecodingFormat, org.apache.flink.table.connector.format.DecodingFormat, [I, [I, java.lang.String, java.util.List, java.util.regex.Pattern, java.util.Properties, org.apache.flink.streaming.connectors.kafka.config.StartupMode, java.util.Map, long, org.apache.flink.streaming.connectors.kafka.config.BoundedMode, java.util.Map, long, boolean, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaDynamicSource.java:216)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSource.<init>(org.apache.flink.table.types.DataType, org.apache.flink.table.connector.format.DecodingFormat, org.apache.flink.table.connector.format.DecodingFormat, [I, [I, java.lang.String, java.util.List, java.util.regex.Pattern, java.util.Properties, org.apache.flink.streaming.connectors.kafka.config.StartupMode, java.util.Map, long, org.apache.flink.streaming.connectors.kafka.config.BoundedMode, java.util.Map, long, boolean, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaDynamicSource.java:218)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSource.<init>(org.apache.flink.table.types.DataType, org.apache.flink.table.connector.format.DecodingFormat, org.apache.flink.table.connector.format.DecodingFormat, [I, [I, java.lang.String, java.util.List, java.util.regex.Pattern, java.util.Properties, org.apache.flink.streaming.connectors.kafka.config.StartupMode, java.util.Map, long, org.apache.flink.streaming.connectors.kafka.config.BoundedMode, java.util.Map, long, boolean, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaDynamicSource.java:220)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSource.<init>(org.apache.flink.table.types.DataType, org.apache.flink.table.connector.format.DecodingFormat, org.apache.flink.table.connector.format.DecodingFormat, [I, [I, java.lang.String, java.util.List, java.util.regex.Pattern, java.util.Properties, org.apache.flink.streaming.connectors.kafka.config.StartupMode, java.util.Map, long, org.apache.flink.streaming.connectors.kafka.config.BoundedMode, java.util.Map, long, boolean, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaDynamicSource.java:224)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSource.<init>(org.apache.flink.table.types.DataType, org.apache.flink.table.connector.format.DecodingFormat, org.apache.flink.table.connector.format.DecodingFormat, [I, [I, java.lang.String, java.util.List, java.util.regex.Pattern, java.util.Properties, org.apache.flink.streaming.connectors.kafka.config.StartupMode, java.util.Map, long, org.apache.flink.streaming.connectors.kafka.config.BoundedMode, java.util.Map, long, boolean, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaDynamicSource.java:226)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.ReducingUpsertWriter.<init>(org.apache.flink.connector.kafka.sink.TwoPhaseCommittingStatefulSink$PrecommittingStatefulSinkWriter, org.apache.flink.table.types.DataType, [I, org.apache.flink.streaming.connectors.kafka.table.SinkBufferFlushMode, org.apache.flink.api.common.operators.ProcessingTimeService, java.util.function.Function)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean)> in (ReducingUpsertWriter.java:70)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.ReducingUpsertWriter.<init>(org.apache.flink.connector.kafka.sink.TwoPhaseCommittingStatefulSink$PrecommittingStatefulSinkWriter, org.apache.flink.table.types.DataType, [I, org.apache.flink.streaming.connectors.kafka.table.SinkBufferFlushMode, org.apache.flink.api.common.operators.ProcessingTimeService, java.util.function.Function)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (ReducingUpsertWriter.java:71)
-Constructor <org.apache.flink.streaming.connectors.kafka.table.ReducingUpsertWriter.<init>(org.apache.flink.connector.kafka.sink.TwoPhaseCommittingStatefulSink$PrecommittingStatefulSinkWriter, org.apache.flink.table.types.DataType, [I, org.apache.flink.streaming.connectors.kafka.table.SinkBufferFlushMode, org.apache.flink.api.common.operators.ProcessingTimeService, java.util.function.Function)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (ReducingUpsertWriter.java:72)
-Method <org.apache.flink.connector.kafka.dynamic.source.DynamicKafkaSourceBuilder.sanityCheck()> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (DynamicKafkaSourceBuilder.java:291)
-Method <org.apache.flink.connector.kafka.dynamic.source.DynamicKafkaSourceBuilder.sanityCheck()> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (DynamicKafkaSourceBuilder.java:293)
-Method <org.apache.flink.connector.kafka.dynamic.source.DynamicKafkaSourceBuilder.sanityCheck()> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (DynamicKafkaSourceBuilder.java:295)
-Method <org.apache.flink.connector.kafka.dynamic.source.DynamicKafkaSourceBuilder.sanityCheck()> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (DynamicKafkaSourceBuilder.java:299)
-Method <org.apache.flink.connector.kafka.dynamic.source.DynamicKafkaSourceBuilder.setKafkaStreamSubscriber(org.apache.flink.connector.kafka.dynamic.source.enumerator.subscriber.KafkaStreamSubscriber)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (DynamicKafkaSourceBuilder.java:100)
-Method <org.apache.flink.connector.kafka.dynamic.source.DynamicKafkaSourceBuilder.setStreamIds(java.util.Set)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (DynamicKafkaSourceBuilder.java:73)
-Method <org.apache.flink.connector.kafka.dynamic.source.DynamicKafkaSourceBuilder.setStreamPattern(java.util.regex.Pattern)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (DynamicKafkaSourceBuilder.java:86)
-Method <org.apache.flink.connector.kafka.dynamic.source.enumerator.DynamicKafkaSourceEnumStateSerializer.serialize(java.util.Set, java.io.DataOutputStream)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (DynamicKafkaSourceEnumStateSerializer.java:142)
-Method <org.apache.flink.connector.kafka.dynamic.source.enumerator.DynamicKafkaSourceEnumerator.handleSourceEvent(int, org.apache.flink.api.connector.source.SourceEvent)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (DynamicKafkaSourceEnumerator.java:514)
-Method <org.apache.flink.connector.kafka.dynamic.source.enumerator.StoppableKafkaEnumContextProxy.wrapCallAsyncCallable(java.util.concurrent.Callable)> calls method <org.apache.flink.util.ExceptionUtils.findThrowable(java.lang.Throwable, java.lang.Class)> in (StoppableKafkaEnumContextProxy.java:237)
-Method <org.apache.flink.connector.kafka.dynamic.source.enumerator.StoppableKafkaEnumContextProxy.wrapCallAsyncCallableHandler(java.util.function.BiConsumer)> calls method <org.apache.flink.util.ExceptionUtils.findThrowable(java.lang.Throwable, java.lang.Class)> in (StoppableKafkaEnumContextProxy.java:259)
-Method <org.apache.flink.connector.kafka.dynamic.source.reader.DynamicKafkaSourceReader.createReader(java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String, [Ljava.lang.Object;)> in (DynamicKafkaSourceReader.java:418)
-Method <org.apache.flink.connector.kafka.dynamic.source.reader.DynamicKafkaSourceReader.handleSourceEvents(org.apache.flink.api.connector.source.SourceEvent)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (DynamicKafkaSourceReader.java:228)
-Method <org.apache.flink.connector.kafka.sink.FlinkKafkaInternalProducer.abortTransaction()> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (FlinkKafkaInternalProducer.java:100)
-Method <org.apache.flink.connector.kafka.sink.FlinkKafkaInternalProducer.commitTransaction()> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (FlinkKafkaInternalProducer.java:109)
-Method <org.apache.flink.connector.kafka.sink.FlinkKafkaInternalProducer.resumeTransaction(long, short)> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.String, [Ljava.lang.Object;)> in (FlinkKafkaInternalProducer.java:296)
-Method <org.apache.flink.connector.kafka.sink.FlinkKafkaInternalProducer.resumeTransaction(long, short)> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.String, [Ljava.lang.Object;)> in (FlinkKafkaInternalProducer.java:297)
-Method <org.apache.flink.connector.kafka.sink.FlinkKafkaInternalProducer.setTransactionId(java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (FlinkKafkaInternalProducer.java:174)
-Method <org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchemaBuilder.build()> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (KafkaRecordSerializationSchemaBuilder.java:268)
-Method <org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchemaBuilder.build()> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (KafkaRecordSerializationSchemaBuilder.java:269)
-Method <org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchemaBuilder.checkKeySerializerNotSet()> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (KafkaRecordSerializationSchemaBuilder.java:283)
-Method <org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchemaBuilder.checkValueSerializerNotSet()> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (KafkaRecordSerializationSchemaBuilder.java:279)
-Method <org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchemaBuilder.setHeaderProvider(org.apache.flink.connector.kafka.sink.HeaderProvider)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (KafkaRecordSerializationSchemaBuilder.java:218)
-Method <org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchemaBuilder.setKeySerializationSchema(org.apache.flink.api.common.serialization.SerializationSchema)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (KafkaRecordSerializationSchemaBuilder.java:154)
-Method <org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchemaBuilder.setPartitioner(org.apache.flink.connector.kafka.sink.KafkaPartitioner)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (KafkaRecordSerializationSchemaBuilder.java:112)
-Method <org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchemaBuilder.setPartitioner(org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (KafkaRecordSerializationSchemaBuilder.java:99)
-Method <org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchemaBuilder.setTopic(java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (KafkaRecordSerializationSchemaBuilder.java:124)
-Method <org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchemaBuilder.setTopic(java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (KafkaRecordSerializationSchemaBuilder.java:123)
-Method <org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchemaBuilder.setTopicSelector(org.apache.flink.connector.kafka.sink.TopicSelector)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (KafkaRecordSerializationSchemaBuilder.java:139)
-Method <org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchemaBuilder.setTopicSelector(org.apache.flink.connector.kafka.sink.TopicSelector)> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (KafkaRecordSerializationSchemaBuilder.java:137)
-Method <org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchemaBuilder.setValueSerializationSchema(org.apache.flink.api.common.serialization.SerializationSchema)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (KafkaRecordSerializationSchemaBuilder.java:204)
-Method <org.apache.flink.connector.kafka.sink.KafkaSerializerWrapper.open(org.apache.flink.api.common.serialization.SerializationSchema$InitializationContext)> calls method <org.apache.flink.util.InstantiationUtil.instantiate(java.lang.String, java.lang.Class, java.lang.ClassLoader)> in (KafkaSerializerWrapper.java:71)
-Method <org.apache.flink.connector.kafka.sink.KafkaSerializerWrapper.serialize(java.lang.Object)> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (KafkaSerializerWrapper.java:88)
-Method <org.apache.flink.connector.kafka.sink.KafkaSinkBuilder.sanityCheck()> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaSinkBuilder.java:194)
-Method <org.apache.flink.connector.kafka.sink.KafkaSinkBuilder.sanityCheck()> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaSinkBuilder.java:202)
-Method <org.apache.flink.connector.kafka.sink.KafkaSinkBuilder.sanityCheck()> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (KafkaSinkBuilder.java:198)
-Method <org.apache.flink.connector.kafka.sink.KafkaSinkBuilder.setDeliverGuarantee(org.apache.flink.connector.base.DeliveryGuarantee)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaSinkBuilder.java:111)
-Method <org.apache.flink.connector.kafka.sink.KafkaSinkBuilder.setDeliveryGuarantee(org.apache.flink.connector.base.DeliveryGuarantee)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaSinkBuilder.java:97)
-Method <org.apache.flink.connector.kafka.sink.KafkaSinkBuilder.setKafkaProducerConfig(java.util.Properties)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (KafkaSinkBuilder.java:123)
-Method <org.apache.flink.connector.kafka.sink.KafkaSinkBuilder.setProperty(java.lang.String, java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (KafkaSinkBuilder.java:133)
-Method <org.apache.flink.connector.kafka.sink.KafkaSinkBuilder.setRecordSerializer(org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaSinkBuilder.java:151)
-Method <org.apache.flink.connector.kafka.sink.KafkaSinkBuilder.setTransactionalIdPrefix(java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaSinkBuilder.java:175)
-Method <org.apache.flink.connector.kafka.sink.KafkaSinkBuilder.setTransactionalIdPrefix(java.lang.String)> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (KafkaSinkBuilder.java:176)
-Method <org.apache.flink.connector.kafka.sink.KafkaWriter.close()> calls method <org.apache.flink.util.IOUtils.closeAll([Ljava.lang.AutoCloseable;)> in (KafkaWriter.java:244)
-Method <org.apache.flink.connector.kafka.sink.KafkaWriter.close()> calls method <org.apache.flink.util.IOUtils.closeAll(java.lang.Iterable)> in (KafkaWriter.java:245)
-Method <org.apache.flink.connector.kafka.sink.KafkaWriter.close()> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (KafkaWriter.java:246)
-Method <org.apache.flink.connector.kafka.sink.KafkaWriter.getTransactionalProducer(long)> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.String, [Ljava.lang.Object;)> in (KafkaWriter.java:311)
-Method <org.apache.flink.connector.kafka.sink.Recyclable.getObject()> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (Recyclable.java:36)
-Method <org.apache.flink.connector.kafka.source.KafkaSourceBuilder.sanityCheck()> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaSourceBuilder.java:513)
-Method <org.apache.flink.connector.kafka.source.KafkaSourceBuilder.sanityCheck()> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaSourceBuilder.java:518)
-Method <org.apache.flink.connector.kafka.source.KafkaSourceBuilder.sanityCheck()> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaSourceBuilder.java:522)
-Method <org.apache.flink.connector.kafka.source.KafkaSourceBuilder.sanityCheck()> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (KafkaSourceBuilder.java:524)
-Method <org.apache.flink.connector.kafka.source.KafkaSourceBuilder.setKafkaSubscriber(org.apache.flink.connector.kafka.source.enumerator.subscriber.KafkaSubscriber)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (KafkaSourceBuilder.java:203)
-Method <org.apache.flink.connector.kafka.source.enumerator.initializer.ReaderHandledOffsetsInitializer.validate(java.util.Properties)> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (ReaderHandledOffsetsInitializer.java:75)
-Method <org.apache.flink.connector.kafka.source.enumerator.initializer.SpecifiedOffsetsInitializer.validate(java.util.Properties)> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (SpecifiedOffsetsInitializer.java:105)
-Method <org.apache.flink.connector.kafka.source.reader.KafkaPartitionSplitReader$KafkaPartitionSplitRecords.nextRecordFromSplit()> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaPartitionSplitReader.java:540)
-Method <org.apache.flink.connector.kafka.source.reader.KafkaPartitionSplitReader.acquireAndSetStoppingOffsets(java.util.List, java.util.Set)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (KafkaPartitionSplitReader.java:359)
-Method <org.apache.flink.connector.kafka.source.reader.deserializer.KafkaValueOnlyDeserializerWrapper.open(org.apache.flink.api.common.serialization.DeserializationSchema$InitializationContext)> calls method <org.apache.flink.util.InstantiationUtil.instantiate(java.lang.String, java.lang.Class, java.lang.ClassLoader)> in (KafkaValueOnlyDeserializerWrapper.java:65)
-Method <org.apache.flink.streaming.connectors.kafka.internals.KafkaShuffleFetcher$WatermarkHandler.checkAndGetNewWatermark(org.apache.flink.streaming.connectors.kafka.internals.KafkaShuffleFetcher$KafkaShuffleWatermark)> calls method <org.apache.flink.util.Preconditions.checkState(boolean, java.lang.Object)> in (KafkaShuffleFetcher.java:280)
-Method <org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptionsUtil.createKeyFormatProjection(org.apache.flink.configuration.ReadableConfig, org.apache.flink.table.types.DataType)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (KafkaConnectorOptionsUtil.java:500)
-Method <org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptionsUtil.createValueFormatProjection(org.apache.flink.configuration.ReadableConfig, org.apache.flink.table.types.DataType)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (KafkaConnectorOptionsUtil.java:567)
-Method <org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptionsUtil.initializePartitioner(java.lang.String, java.lang.ClassLoader)> calls method <org.apache.flink.util.InstantiationUtil.instantiate(java.lang.String, java.lang.Class, java.lang.ClassLoader)> in (KafkaConnectorOptionsUtil.java:480)
-Method <org.apache.flink.streaming.connectors.kafka.table.ReducingUpsertWriter$WrappedContext.currentWatermark()> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (ReducingUpsertWriter.java:177)
diff --git a/flink-connector-kafka/archunit-violations/d853eb69-8c04-4246-9a5e-4f5911286b1d b/flink-connector-kafka/archunit-violations/d853eb69-8c04-4246-9a5e-4f5911286b1d
index dfb76a0c5..fddc6df33 100644
--- a/flink-connector-kafka/archunit-violations/d853eb69-8c04-4246-9a5e-4f5911286b1d
+++ b/flink-connector-kafka/archunit-violations/d853eb69-8c04-4246-9a5e-4f5911286b1d
@@ -2,4 +2,3 @@ org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema.open(org.ap
 org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema.serialize(java.lang.Object, org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema$KafkaSinkContext, java.lang.Long): Argument leaf type org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema$KafkaSinkContext does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
 org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer.getPartitionOffsets(java.util.Collection, org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer$PartitionOffsetsRetriever): Argument leaf type org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer$PartitionOffsetsRetriever does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
 org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.invoke(org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer$KafkaTransactionState, java.lang.Object, org.apache.flink.streaming.api.functions.sink.SinkFunction$Context): Argument leaf type org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer$KafkaTransactionState does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
-org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition.dropLeaderData(java.util.List): Argument leaf type org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartitionLeader does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/FlinkKafkaInternalProducer.java b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/FlinkKafkaInternalProducer.java
index 6e618cbe0..0469c0707 100644
--- a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/FlinkKafkaInternalProducer.java
+++ b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/FlinkKafkaInternalProducer.java
@@ -37,6 +37,7 @@
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.errors.ProducerFencedException;
 import org.apache.kafka.common.requests.FindCoordinatorRequest;
 import org.slf4j.Logger;
@@ -156,6 +157,11 @@ public List<PartitionInfo> partitionsFor(String topic) {
         return kafkaProducer.metrics();
     }
 
+    @Override
+    public Uuid clientInstanceId(Duration timeout) {
+        return kafkaProducer.clientInstanceId(timeout);
+    }
+
     @Override
     public void close() {
         throw new UnsupportedOperationException(
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/testutils/DockerImageVersions.java b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/testutils/DockerImageVersions.java
index 2d262963b..0f40dc26f 100644
--- a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/testutils/DockerImageVersions.java
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/testutils/DockerImageVersions.java
@@ -24,9 +24,9 @@
  */
 public class DockerImageVersions {
 
-    public static final String KAFKA = "confluentinc/cp-kafka:7.4.4";
+    public static final String KAFKA = "confluentinc/cp-kafka:7.7.2";
 
-    public static final String SCHEMA_REGISTRY = "confluentinc/cp-schema-registry:7.4.4";
+    public static final String SCHEMA_REGISTRY = "confluentinc/cp-schema-registry:7.7.2";
 
-    public static final String ZOOKEEPER = "zookeeper:3.4.14";
+    public static final String ZOOKEEPER = "zookeeper:3.8.4";
 }
diff --git a/flink-sql-connector-kafka/src/main/resources/META-INF/NOTICE b/flink-sql-connector-kafka/src/main/resources/META-INF/NOTICE
index 926976da9..1b9dc9598 100644
--- a/flink-sql-connector-kafka/src/main/resources/META-INF/NOTICE
+++ b/flink-sql-connector-kafka/src/main/resources/META-INF/NOTICE
@@ -6,4 +6,4 @@ The Apache Software Foundation (http://www.apache.org/).
 
 This project bundles the following dependencies under the Apache Software License 2.0. (http://www.apache.org/licenses/LICENSE-2.0.txt)
 
-- org.apache.kafka:kafka-clients:3.4.0
+- org.apache.kafka:kafka-clients:3.9.0
diff --git a/pom.xml b/pom.xml
index 41d9aefc4..99abbe1f6 100644
--- a/pom.xml
+++ b/pom.xml
@@ -51,10 +51,10 @@ under the License.
 
     <properties>
         <flink.version>1.20.0</flink.version>
-        <kafka.version>3.4.0</kafka.version>
-        <confluent.version>7.4.4</confluent.version>
+        <kafka.version>3.9.0</kafka.version>
+        <confluent.version>7.7.1</confluent.version>
 
-        <jackson-bom.version>2.15.2</jackson-bom.version>
+        <jackson-bom.version>2.16.2</jackson-bom.version>
         <junit4.version>4.13.2</junit4.version>
         <junit5.version>5.9.1</junit5.version>
         <assertj.version>3.23.1</assertj.version>
@@ -65,8 +65,8 @@ under the License.
         <byte-buddy.version>1.12.10</byte-buddy.version>
         <commons-cli.version>1.5.0</commons-cli.version>
         <scala.binary.version>2.12</scala.binary.version>
-        <scala-reflect.version>2.12.7</scala-reflect.version>
-        <scala-library.version>2.12.7</scala-library.version>
+        <scala-reflect.version>2.12.19</scala-reflect.version>
+        <scala-library.version>2.12.19</scala-library.version>
         <snappy-java.version>1.1.10.5</snappy-java.version>
         <avro.version>1.11.4</avro.version>
         <guava.version>32.1.2-jre</guava.version>
@@ -74,7 +74,7 @@ under the License.
         <japicmp.skip>false</japicmp.skip>
         <japicmp.referenceVersion>1.17.0</japicmp.referenceVersion>
 
-        <slf4j.version>1.7.32</slf4j.version>
+        <slf4j.version>1.7.36</slf4j.version>
         <log4j.version>2.17.1</log4j.version>
 
         <flink.parent.artifactId>flink-connector-kafka-parent</flink.parent.artifactId>