Merge branch 'main' into feature/kafka4

Shekharrajak · web-flow · commit 568b5d3f8776 · 2025-09-02T09:47:51.000+05:30
diff --git a/.github/workflows/push_pr.yml b/.github/workflows/push_pr.yml
@@ -28,7 +28,7 @@ jobs:
   compile_and_test:
     strategy:
       matrix:
-        flink: [ 2.0.0 ]
+        flink: [ 2.1.0 ]
         jdk: [ '11, 17, 21' ]
     uses: apache/flink-connector-shared-utils/.github/workflows/ci.yml@ci_utils
     with:
@@ -37,7 +37,7 @@ jobs:
   python_test:
     strategy:
       matrix:
-        flink: [ 2.0.0 ]
+        flink: [ 2.1.0 ]
         jdk: [ '11, 17, 21' ]
     uses: apache/flink-connector-shared-utils/.github/workflows/python_ci.yml@ci_utils
     with:
diff --git a/.gitignore b/.gitignore
@@ -3,6 +3,9 @@
 scalastyle-output.xml
 .classpath
 .idea/
+.idea/*
+!.idea/vcs.xml
+.vscode
 .metadata
 .settings
 .project
diff --git a/docs/content.zh/docs/connectors/datastream/kafka.md b/docs/content.zh/docs/connectors/datastream/kafka.md
@@ -33,7 +33,7 @@ Flink 提供了 [Apache Kafka](https://kafka.apache.org) 连接器使用精确
 
 Apache Flink 集成了通用的 Kafka 连接器，它会尽力与 Kafka client 的最新版本保持同步。
 该连接器使用的 Kafka client 版本可能会在 Flink 版本之间发生变化。
-当前 Kafka client 向后兼容 0.10.0 或更高版本的 Kafka broker。
+当前 Kafka client 向后兼容 2.1.0 或更高版本的 Kafka broker。
 有关 Kafka 兼容性的更多细节，请参考 [Kafka 官方文档](https://kafka.apache.org/protocol.html#protocol_compatibility)。
 
 {{< connector_artifact flink-connector-kafka kafka >}}
diff --git a/docs/content/docs/connectors/datastream/kafka.md b/docs/content/docs/connectors/datastream/kafka.md
@@ -33,7 +33,7 @@ Flink provides an [Apache Kafka](https://kafka.apache.org) connector for reading
 
 Apache Flink ships with a universal Kafka connector which attempts to track the latest version of the Kafka client.
 The version of the client it uses may change between Flink releases.
-Modern Kafka clients are backwards compatible with broker versions 0.10.0 or later.
+Modern Kafka clients are backwards compatible with broker versions 2.1.0 or later.
 For details on Kafka compatibility, please refer to the official [Kafka documentation](https://kafka.apache.org/protocol.html#protocol_compatibility).
 
 {{< connector_artifact flink-connector-kafka kafka >}}
diff --git a/flink-connector-kafka/.gitignore b/flink-connector-kafka/.gitignore
@@ -0,0 +1 @@
+/latest_offset_resume_topic*
diff --git a/flink-connector-kafka/archunit-violations/c0d94764-76a0-4c50-b617-70b1754c4612 b/flink-connector-kafka/archunit-violations/c0d94764-76a0-4c50-b617-70b1754c4612
@@ -23,8 +23,6 @@ Method <org.apache.flink.connector.kafka.dynamic.source.reader.DynamicKafkaSourc
 Method <org.apache.flink.connector.kafka.dynamic.source.reader.DynamicKafkaSourceReader.syncAvailabilityHelperWithReaders()> calls method <org.apache.flink.streaming.runtime.io.MultipleFuturesAvailabilityHelper.anyOf(int, java.util.concurrent.CompletableFuture)> in (DynamicKafkaSourceReader.java:500)
 Method <org.apache.flink.connector.kafka.sink.ExactlyOnceKafkaWriter.getProducerPool()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (ExactlyOnceKafkaWriter.java:0)
 Method <org.apache.flink.connector.kafka.sink.ExactlyOnceKafkaWriter.getTransactionalIdPrefix()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (ExactlyOnceKafkaWriter.java:0)
-Method <org.apache.flink.connector.kafka.sink.internal.KafkaCommitter.getBackchannel()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (KafkaCommitter.java:0)
-Method <org.apache.flink.connector.kafka.sink.internal.KafkaCommitter.getCommittingProducer()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (KafkaCommitter.java:0)
 Method <org.apache.flink.connector.kafka.sink.KafkaSink.addPostCommitTopology(org.apache.flink.streaming.api.datastream.DataStream)> calls method <org.apache.flink.api.dag.Transformation.getCoLocationGroupKey()> in (KafkaSink.java:178)
 Method <org.apache.flink.connector.kafka.sink.KafkaSink.addPostCommitTopology(org.apache.flink.streaming.api.datastream.DataStream)> calls method <org.apache.flink.api.dag.Transformation.getInputs()> in (KafkaSink.java:181)
 Method <org.apache.flink.connector.kafka.sink.KafkaSink.addPostCommitTopology(org.apache.flink.streaming.api.datastream.DataStream)> calls method <org.apache.flink.api.dag.Transformation.getOutputType()> in (KafkaSink.java:177)
@@ -34,6 +32,8 @@ Method <org.apache.flink.connector.kafka.sink.KafkaSink.addPostCommitTopology(or
 Method <org.apache.flink.connector.kafka.sink.KafkaSink.getKafkaProducerConfig()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (KafkaSink.java:0)
 Method <org.apache.flink.connector.kafka.sink.KafkaSinkBuilder.setRecordSerializer(org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema)> calls method <org.apache.flink.api.java.ClosureCleaner.clean(java.lang.Object, org.apache.flink.api.common.ExecutionConfig$ClosureCleanerLevel, boolean)> in (KafkaSinkBuilder.java:154)
 Method <org.apache.flink.connector.kafka.sink.KafkaWriter.getCurrentProducer()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (KafkaWriter.java:0)
+Method <org.apache.flink.connector.kafka.sink.internal.KafkaCommitter.getBackchannel()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (KafkaCommitter.java:0)
+Method <org.apache.flink.connector.kafka.sink.internal.KafkaCommitter.getCommittingProducer()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (KafkaCommitter.java:0)
 Method <org.apache.flink.connector.kafka.sink.internal.ProducerPoolImpl.getProducers()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (ProducerPoolImpl.java:0)
 Method <org.apache.flink.connector.kafka.source.KafkaSource.createReader(org.apache.flink.api.connector.source.SourceReaderContext, java.util.function.Consumer)> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (KafkaSource.java:0)
 Method <org.apache.flink.connector.kafka.source.KafkaSource.getConfiguration()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (KafkaSource.java:0)
@@ -50,6 +50,4 @@ Method <org.apache.flink.connector.kafka.source.reader.KafkaSourceReader.getOffs
 Method <org.apache.flink.streaming.connectors.kafka.table.DynamicKafkaRecordSerializationSchema.createProjectedRow(org.apache.flink.table.data.RowData, org.apache.flink.types.RowKind, [Lorg.apache.flink.table.data.RowData$FieldGetter;)> has parameter of type <[Lorg.apache.flink.table.data.RowData$FieldGetter;> in (DynamicKafkaRecordSerializationSchema.java:0)
 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.table.types.logical.utils.LogicalTypeChecks.getFieldNames(org.apache.flink.table.types.logical.LogicalType)> in (KafkaConnectorOptionsUtil.java:520)
 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.table.types.logical.utils.LogicalTypeChecks.getFieldCount(org.apache.flink.table.types.logical.LogicalType)> in (KafkaConnectorOptionsUtil.java:564)
-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:408)
 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:574)
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/metadata/SingleClusterTopicMetadataService.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/metadata/SingleClusterTopicMetadataService.java
@@ -72,8 +72,8 @@ public Set<KafkaStream> getAllStreams() {
     @Override
     public Map<String, KafkaStream> describeStreams(Collection<String> streamIds) {
         try {
-            return getAdminClient().describeTopics(new ArrayList<>(streamIds)).allTopicNames().get().keySet()
-                    .stream()
+            return getAdminClient().describeTopics(new ArrayList<>(streamIds)).allTopicNames().get()
+                    .keySet().stream()
                     .collect(Collectors.toMap(topic -> topic, this::createKafkaStream));
         } catch (InterruptedException | ExecutionException e) {
             throw new RuntimeException("Fetching all streams failed", e);
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumerator.java b/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.ListConsumerGroupOffsetsOptions;
 import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsSpec;
 import org.apache.kafka.clients.admin.ListOffsetsResult;
 import org.apache.kafka.clients.admin.OffsetSpec;
@@ -534,15 +533,12 @@ public PartitionOffsetsRetrieverImpl(AdminClient adminClient, String groupId) {
 
         @Override
         public Map<TopicPartition, Long> committedOffsets(Collection<TopicPartition> partitions) {
-            ListConsumerGroupOffsetsSpec groupSpec = 
-                    new ListConsumerGroupOffsetsSpec()
-                            .topicPartitions(new ArrayList<>(partitions));
-            Map<String, ListConsumerGroupOffsetsSpec> groupSpecs = Collections.singletonMap(groupId, groupSpec);
-            ListConsumerGroupOffsetsOptions options = new ListConsumerGroupOffsetsOptions();
+            ListConsumerGroupOffsetsSpec offsetsSpec =
+                    new ListConsumerGroupOffsetsSpec().topicPartitions(partitions);
             try {
                 return adminClient
-                        .listConsumerGroupOffsets(groupSpecs, options)
-                        .all()
+                        .listConsumerGroupOffsets(Collections.singletonMap(groupId, offsetsSpec))
+                        .partitionsToOffsetAndMetadata()
                         .thenApply(
                                 result -> {
                                     Map<TopicPartition, Long> offsets = new HashMap<>();
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSink.java b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSink.java
@@ -44,7 +44,6 @@
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.table.types.logical.LogicalType;
-import org.apache.flink.table.types.utils.DataTypeUtils;
 
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.header.Header;
@@ -405,7 +404,8 @@ private RowData.FieldGetter[] getFieldGetters(
         }
         DataType physicalFormatDataType = Projection.of(projection).project(this.physicalDataType);
         if (prefix != null) {
-            physicalFormatDataType = DataTypeUtils.stripRowPrefix(physicalFormatDataType, prefix);
+            physicalFormatDataType =
+                    TableDataTypeUtils.stripRowPrefix(physicalFormatDataType, prefix);
         }
         return format.createRuntimeEncoder(context, physicalFormatDataType);
     }
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSource.java b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSource.java
@@ -49,7 +49,6 @@
 import org.apache.flink.table.data.StringData;
 import org.apache.flink.table.data.TimestampData;
 import org.apache.flink.table.types.DataType;
-import org.apache.flink.table.types.utils.DataTypeUtils;
 import org.apache.flink.util.Preconditions;
 
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -571,7 +570,8 @@ private KafkaRecordDeserializationSchema<RowData> createKafkaDeserializationSche
         }
         DataType physicalFormatDataType = Projection.of(projection).project(this.physicalDataType);
         if (prefix != null) {
-            physicalFormatDataType = DataTypeUtils.stripRowPrefix(physicalFormatDataType, prefix);
+            physicalFormatDataType =
+                    TableDataTypeUtils.stripRowPrefix(physicalFormatDataType, prefix);
         }
         return format.createRuntimeDecoder(context, physicalFormatDataType);
     }
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/TableDataTypeUtils.java b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/TableDataTypeUtils.java
@@ -0,0 +1,83 @@
+package org.apache.flink.streaming.connectors.kafka.table;
+
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.FieldsDataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.flink.table.types.logical.LogicalTypeRoot.ROW;
+
+/**
+ * Utility class for manipulating {@link DataType} objects, particularly for table schemas. This
+ * class provides methods that were removed from the Flink API in version 2.1.0. See <a
+ * href="https://github.com/apache/flink/pull/26784">Flink PR 26784</a>.
+ */
+public class TableDataTypeUtils {
+
+    protected static final String STRIP_ROW_NO_ROW_ERROR_MSG = "Row data type expected.";
+    protected static final String RENAME_ROW_LENGTH_MISMATCH_ERROR_MSG =
+            "Row length and new names must match.";
+
+    /**
+     * Removes a string prefix from the fields of the given row data type.
+     *
+     * @param dataType The row data type to modify.
+     * @param prefix The prefix to remove from the field names.
+     * @return A new DataType with the modified field names.
+     * @throws IllegalArgumentException if the provided dataType is not of ROW type.
+     */
+    public static DataType stripRowPrefix(DataType dataType, String prefix) {
+
+        if (!dataType.getLogicalType().is(ROW)) {
+            throw new IllegalArgumentException(STRIP_ROW_NO_ROW_ERROR_MSG);
+        }
+
+        final RowType rowType = (RowType) dataType.getLogicalType();
+        final List<String> newFieldNames =
+                rowType.getFieldNames().stream()
+                        .map(
+                                s -> {
+                                    if (s.startsWith(prefix)) {
+                                        return s.substring(prefix.length());
+                                    }
+                                    return s;
+                                })
+                        .collect(Collectors.toList());
+        final LogicalType newRowType = renameRowFields(rowType, newFieldNames);
+        return new FieldsDataType(
+                newRowType, dataType.getConversionClass(), dataType.getChildren());
+    }
+
+    /**
+     * Renames the fields of the given {@link RowType}.
+     *
+     * @param rowType The original RowType.
+     * @param newFieldNames The new field names to apply.
+     * @return A new RowType with the updated field names.
+     * @throws IllegalArgumentException if the number of new field names does not match the number
+     *     of fields in the original RowType.
+     */
+    public static RowType renameRowFields(RowType rowType, List<String> newFieldNames) {
+
+        if (!(rowType.getFieldCount() == newFieldNames.size())) {
+            throw new IllegalArgumentException(RENAME_ROW_LENGTH_MISMATCH_ERROR_MSG);
+        }
+
+        final List<RowType.RowField> newFields =
+                IntStream.range(0, rowType.getFieldCount())
+                        .mapToObj(
+                                pos -> {
+                                    final RowType.RowField oldField = rowType.getFields().get(pos);
+                                    return new RowType.RowField(
+                                            newFieldNames.get(pos),
+                                            oldField.getType(),
+                                            oldField.getDescription().orElse(null));
+                                })
+                        .collect(Collectors.toList());
+        return new RowType(rowType.isNullable(), newFields);
+    }
+}
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterFaultToleranceITCase.java b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterFaultToleranceITCase.java
@@ -40,7 +40,11 @@ public class KafkaWriterFaultToleranceITCase extends KafkaWriterTestBase {
     private static final String INIT_KAFKA_RETRIES = "0";
     private static final String INIT_KAFKA_REQUEST_TIMEOUT_MS = "1000";
     private static final String INIT_KAFKA_MAX_BLOCK_MS = "1000";
-    private static final String INIT_KAFKA_DELIVERY_TIMEOUT_MS = "1000";
+    /**
+     * The delivery timeout has to be greater than the request timeout as the latter is part of the
+     * former and this is enforced by a compile time check.
+     */
+    private static final String INIT_KAFKA_DELIVERY_TIMEOUT_MS = "1500";
 
     @RegisterExtension
     public static final MiniClusterExtension MINI_CLUSTER_RESOURCE =
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/testutils/KafkaSourceTestEnv.java b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/testutils/KafkaSourceTestEnv.java
@@ -22,7 +22,7 @@
 import org.apache.flink.streaming.connectors.kafka.KafkaTestBase;
 
 import org.apache.kafka.clients.admin.AdminClient;
-import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsOptions;
+import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsSpec;
 import org.apache.kafka.clients.admin.RecordsToDelete;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
@@ -36,6 +36,7 @@
 
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -238,10 +239,10 @@ public static void setupCommittedOffsets(String topic)
         Map<TopicPartition, OffsetAndMetadata> toVerify =
                 adminClient
                         .listConsumerGroupOffsets(
-                                GROUP_ID,
-                                new ListConsumerGroupOffsetsOptions()
-                                        .topicPartitions(
-                                                new ArrayList<>(committedOffsets.keySet())))
+                                Collections.singletonMap(
+                                        GROUP_ID,
+                                        new ListConsumerGroupOffsetsSpec()
+                                                .topicPartitions(committedOffsets.keySet())))
                         .partitionsToOffsetAndMetadata()
                         .get();
         assertThat(toVerify).as("The offsets are not committed").isEqualTo(committedOffsets);
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
@@ -282,8 +282,13 @@ public KafkaOffsetHandlerImpl() {
 
         @Override
         public Long getCommittedOffset(String topicName, int partition) {
-            OffsetAndMetadata committed =
-                    offsetClient.committed(new TopicPartition(topicName, partition));
+            TopicPartition topicPartition = new TopicPartition(topicName, partition);
+            Map<TopicPartition, OffsetAndMetadata> committedMap =
+                    offsetClient.committed(Collections.singleton(topicPartition));
+            if (committedMap == null) {
+                return null;
+            }
+            OffsetAndMetadata committed = committedMap.get(topicPartition);
             return (committed != null) ? committed.offset() : null;
         }
 
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/TableDataTypeUtilsTest.java b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/TableDataTypeUtilsTest.java
diff --git a/flink-python/tox.ini b/flink-python/tox.ini
diff --git a/flink-sql-connector-kafka/src/main/resources/META-INF/NOTICE b/flink-sql-connector-kafka/src/main/resources/META-INF/NOTICE
diff --git a/tools/ci/log4j.properties b/tools/ci/log4j.properties

Original file line number	Diff line number	Diff line change
`@@ -44,7 +44,6 @@`
`44`	`44`	`import org.apache.flink.table.data.RowData;`
`45`	`45`	`import org.apache.flink.table.types.DataType;`
`46`	`46`	`import org.apache.flink.table.types.logical.LogicalType;`
`47`		`-import org.apache.flink.table.types.utils.DataTypeUtils;`
`48`	`47`
`49`	`48`	`import org.apache.kafka.clients.producer.ProducerConfig;`
`50`	`49`	`import org.apache.kafka.common.header.Header;`
`@@ -405,7 +404,8 @@ private RowData.FieldGetter[] getFieldGetters(`
`405`	`404`	`}`
`406`	`405`	`DataType physicalFormatDataType = Projection.of(projection).project(this.physicalDataType);`
`407`	`406`	`if (prefix != null) {`
`408`		`- physicalFormatDataType = DataTypeUtils.stripRowPrefix(physicalFormatDataType, prefix);`
	`407`	`+ physicalFormatDataType =`
	`408`	`+ TableDataTypeUtils.stripRowPrefix(physicalFormatDataType, prefix);`
`409`	`409`	`}`
`410`	`410`	`return format.createRuntimeEncoder(context, physicalFormatDataType);`
`411`	`411`	`}`
Original file line number	Diff line number	Diff line change
`@@ -49,7 +49,6 @@`
`49`	`49`	`import org.apache.flink.table.data.StringData;`
`50`	`50`	`import org.apache.flink.table.data.TimestampData;`
`51`	`51`	`import org.apache.flink.table.types.DataType;`
`52`		`-import org.apache.flink.table.types.utils.DataTypeUtils;`
`53`	`52`	`import org.apache.flink.util.Preconditions;`
`54`	`53`
`55`	`54`	`import org.apache.kafka.clients.consumer.ConsumerConfig;`
`@@ -571,7 +570,8 @@ private KafkaRecordDeserializationSchema<RowData> createKafkaDeserializationSche`
`571`	`570`	`}`
`572`	`571`	`DataType physicalFormatDataType = Projection.of(projection).project(this.physicalDataType);`
`573`	`572`	`if (prefix != null) {`
`574`		`- physicalFormatDataType = DataTypeUtils.stripRowPrefix(physicalFormatDataType, prefix);`
	`573`	`+ physicalFormatDataType =`
	`574`	`+ TableDataTypeUtils.stripRowPrefix(physicalFormatDataType, prefix);`
`575`	`575`	`}`
`576`	`576`	`return format.createRuntimeDecoder(context, physicalFormatDataType);`
`577`	`577`	`}`