You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by di...@apache.org on 2024/09/25 02:55:02 UTC
(flink) 01/03: [FLINK-36311][format/avro] Remove deprecated APIs in Flink Avro format
This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit de40ebe53797a4abe023a437e63d9ba5aa095d40
Author: Dian Fu <di...@apache.org>
AuthorDate: Wed Sep 18 14:16:20 2024 +0800
[FLINK-36311][format/avro] Remove deprecated APIs in Flink Avro format
---
.../avro/typeutils/AvroSchemaConverter.java | 6 +-
.../formats/avro/typeutils/AvroSerializer.java | 18 -
.../avro/AvroRowDeSerializationSchemaTest.java | 374 ---------------------
flink-python/pom.xml | 10 +
.../formats/avro/AvroRowDeserializationSchema.java | 9 +-
.../formats/avro/AvroRowSerializationSchema.java | 7 +-
6 files changed, 17 insertions(+), 407 deletions(-)
diff --git a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSchemaConverter.java b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSchemaConverter.java
index 178056e145a..651bda15053 100644
--- a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSchemaConverter.java
+++ b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSchemaConverter.java
@@ -21,8 +21,8 @@ package org.apache.flink.formats.avro.typeutils;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
-import org.apache.flink.formats.avro.AvroRowDeserializationSchema;
-import org.apache.flink.formats.avro.AvroRowSerializationSchema;
+import org.apache.flink.formats.avro.AvroRowDataDeserializationSchema;
+import org.apache.flink.formats.avro.AvroRowDataSerializationSchema;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.types.AtomicDataType;
import org.apache.flink.table.types.DataType;
@@ -56,7 +56,7 @@ import java.util.List;
* SQL API.
*
* <p>Note: Changes in this class need to be kept in sync with the corresponding runtime classes
- * {@link AvroRowDeserializationSchema} and {@link AvroRowSerializationSchema}.
+ * {@link AvroRowDataDeserializationSchema} and {@link AvroRowDataSerializationSchema}.
*/
public class AvroSchemaConverter {
diff --git a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSerializer.java b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSerializer.java
index 99e9de566c6..69e9073d9b1 100644
--- a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSerializer.java
+++ b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSerializer.java
@@ -137,13 +137,6 @@ public class AvroSerializer<T> extends TypeSerializer<T> {
this.previousSchema = checkNotNull(previousSchema);
}
- /** @deprecated Use {@link AvroSerializer#AvroSerializer(Class)} instead. */
- @Deprecated
- @SuppressWarnings("unused")
- public AvroSerializer(Class<T> type, Class<? extends T> typeToInstantiate) {
- this(type);
- }
-
// ------------------------------------------------------------------------
@Nonnull
@@ -367,17 +360,6 @@ public class AvroSerializer<T> extends TypeSerializer<T> {
// Serializer Snapshots
// ------------------------------------------------------------------------
- /**
- * A config snapshot for the Avro Serializer that stores the Avro Schema to check compatibility.
- * This class is now deprecated and only kept for backward comparability.
- */
- @Deprecated
- public static final class AvroSchemaSerializerConfigSnapshot<T>
- extends AvroSerializerSnapshot<T> {
-
- public AvroSchemaSerializerConfigSnapshot() {}
- }
-
// -------- backwards compatibility with 1.5, 1.6 -----------
private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
diff --git a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroRowDeSerializationSchemaTest.java b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroRowDeSerializationSchemaTest.java
deleted file mode 100644
index 1349ca2e03e..00000000000
--- a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroRowDeSerializationSchemaTest.java
+++ /dev/null
@@ -1,374 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.formats.avro;
-
-import org.apache.flink.api.java.tuple.Tuple3;
-import org.apache.flink.api.java.tuple.Tuple4;
-import org.apache.flink.formats.avro.utils.AvroTestUtils;
-import org.apache.flink.types.Row;
-import org.apache.flink.util.InstantiationUtil;
-
-import org.apache.avro.Schema;
-import org.apache.avro.generic.GenericRecord;
-import org.apache.avro.specific.SpecificRecord;
-import org.junit.Test;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.ValueSource;
-
-import java.io.IOException;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-
-/** Test for the Avro serialization and deserialization schema. */
-class AvroRowDeSerializationSchemaTest {
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- void testSpecificSerializeDeserializeFromSchema(boolean legacyTimestampMapping)
- throws IOException {
- final Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> testData =
- AvroTestUtils.getSpecificTestData();
- final String schemaString = testData.f1.getSchema().toString();
-
- final AvroRowSerializationSchema serializationSchema =
- new AvroRowSerializationSchema(schemaString);
- final AvroRowDeserializationSchema deserializationSchema =
- new AvroRowDeserializationSchema(schemaString);
-
- if (legacyTimestampMapping) {
- final byte[] bytes = serializationSchema.serialize(testData.f2);
- final Row actual = deserializationSchema.deserialize(bytes);
- assertThat(actual).isEqualTo(testData.f2);
- } else {
- final byte[] bytes = serializationSchema.serialize(testData.f2, false);
- final Row actual = deserializationSchema.deserialize(bytes, false);
- assertThat(actual).isEqualTo(testData.f2);
- }
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- void testGenericSerializeDeserialize(boolean legacyTimestampMapping) throws IOException {
- final Tuple3<GenericRecord, Row, Schema> testData = AvroTestUtils.getGenericTestData();
-
- final AvroRowSerializationSchema serializationSchema =
- new AvroRowSerializationSchema(testData.f2.toString());
- final AvroRowDeserializationSchema deserializationSchema =
- new AvroRowDeserializationSchema(testData.f2.toString());
-
- if (legacyTimestampMapping) {
- final byte[] bytes = serializationSchema.serialize(testData.f1);
- final Row actual = deserializationSchema.deserialize(bytes);
- assertThat(actual).isEqualTo(testData.f1);
- } else {
- final byte[] bytes = serializationSchema.serialize(testData.f1, false);
- final Row actual = deserializationSchema.deserialize(bytes, false);
- assertThat(actual).isEqualTo(testData.f1);
- }
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- void testSpecificSerializeFromClassSeveralTimes(boolean legacyTimestampMapping)
- throws IOException {
- final Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> testData =
- AvroTestUtils.getSpecificTestData();
-
- final AvroRowSerializationSchema serializationSchema =
- new AvroRowSerializationSchema(testData.f0);
- final AvroRowDeserializationSchema deserializationSchema =
- new AvroRowDeserializationSchema(testData.f0);
-
- if (legacyTimestampMapping) {
- serializationSchema.serialize(testData.f2);
- serializationSchema.serialize(testData.f2);
- final byte[] bytes = serializationSchema.serialize(testData.f2);
- final Row actual = deserializationSchema.deserialize(bytes);
- assertThat(actual).isEqualTo(testData.f2);
- } else {
- serializationSchema.serialize(testData.f2, false);
- serializationSchema.serialize(testData.f2, false);
- final byte[] bytes = serializationSchema.serialize(testData.f2, false);
- final Row actual = deserializationSchema.deserialize(bytes, false);
- assertThat(actual).isEqualTo(testData.f2);
- }
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- void testSpecificSerializeFromSchemaSeveralTimes(boolean legacyTimestampMapping)
- throws IOException {
- final Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> testData =
- AvroTestUtils.getSpecificTestData();
- final String schemaString = testData.f1.getSchema().toString();
-
- final AvroRowSerializationSchema serializationSchema =
- new AvroRowSerializationSchema(schemaString);
- final AvroRowDeserializationSchema deserializationSchema =
- new AvroRowDeserializationSchema(schemaString);
-
- if (legacyTimestampMapping) {
- serializationSchema.serialize(testData.f2);
- serializationSchema.serialize(testData.f2);
- final byte[] bytes = serializationSchema.serialize(testData.f2);
- final Row actual = deserializationSchema.deserialize(bytes);
- assertThat(actual).isEqualTo(testData.f2);
- } else {
- serializationSchema.serialize(testData.f2, false);
- serializationSchema.serialize(testData.f2, false);
- final byte[] bytes = serializationSchema.serialize(testData.f2, false);
- final Row actual = deserializationSchema.deserialize(bytes, false);
- assertThat(actual).isEqualTo(testData.f2);
- }
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- void testGenericSerializeSeveralTimes(boolean legacyTimestampMapping) throws IOException {
- final Tuple3<GenericRecord, Row, Schema> testData = AvroTestUtils.getGenericTestData();
-
- final AvroRowSerializationSchema serializationSchema =
- new AvroRowSerializationSchema(testData.f2.toString());
- final AvroRowDeserializationSchema deserializationSchema =
- new AvroRowDeserializationSchema(testData.f2.toString());
-
- if (legacyTimestampMapping) {
- serializationSchema.serialize(testData.f1);
- serializationSchema.serialize(testData.f1);
- final byte[] bytes = serializationSchema.serialize(testData.f1);
- final Row actual = deserializationSchema.deserialize(bytes);
- assertThat(actual).isEqualTo(testData.f1);
- } else {
- serializationSchema.serialize(testData.f1, false);
- serializationSchema.serialize(testData.f1, false);
- final byte[] bytes = serializationSchema.serialize(testData.f1, false);
- final Row actual = deserializationSchema.deserialize(bytes, false);
- assertThat(actual).isEqualTo(testData.f1);
- }
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- void testSpecificDeserializeFromClassSeveralTimes(boolean legacyTimestampMapping)
- throws IOException {
- final Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> testData =
- AvroTestUtils.getSpecificTestData();
-
- final AvroRowSerializationSchema serializationSchema =
- new AvroRowSerializationSchema(testData.f0);
- final AvroRowDeserializationSchema deserializationSchema =
- new AvroRowDeserializationSchema(testData.f0);
-
- if (legacyTimestampMapping) {
- final byte[] bytes = serializationSchema.serialize(testData.f2);
- deserializationSchema.deserialize(bytes);
- deserializationSchema.deserialize(bytes);
- final Row actual = deserializationSchema.deserialize(bytes);
- assertThat(actual).isEqualTo(testData.f2);
- } else {
- final byte[] bytes = serializationSchema.serialize(testData.f2, false);
- deserializationSchema.deserialize(bytes, false);
- deserializationSchema.deserialize(bytes, false);
- final Row actual = deserializationSchema.deserialize(bytes, false);
- assertThat(actual).isEqualTo(testData.f2);
- }
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- void testSpecificDeserializeFromSchemaSeveralTimes(boolean legacyTimestampMapping)
- throws IOException {
- final Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> testData =
- AvroTestUtils.getSpecificTestData();
- final String schemaString = testData.f1.getSchema().toString();
-
- final AvroRowSerializationSchema serializationSchema =
- new AvroRowSerializationSchema(schemaString);
- final AvroRowDeserializationSchema deserializationSchema =
- new AvroRowDeserializationSchema(schemaString);
-
- if (legacyTimestampMapping) {
- final byte[] bytes = serializationSchema.serialize(testData.f2);
- deserializationSchema.deserialize(bytes);
- deserializationSchema.deserialize(bytes);
- final Row actual = deserializationSchema.deserialize(bytes);
- assertThat(actual).isEqualTo(testData.f2);
- } else {
- final byte[] bytes = serializationSchema.serialize(testData.f2, false);
- deserializationSchema.deserialize(bytes, false);
- deserializationSchema.deserialize(bytes, false);
- final Row actual = deserializationSchema.deserialize(bytes, false);
- assertThat(actual).isEqualTo(testData.f2);
- }
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- void testGenericDeserializeSeveralTimes(boolean legacyTimestampMapping) throws IOException {
- final Tuple3<GenericRecord, Row, Schema> testData = AvroTestUtils.getGenericTestData();
-
- final AvroRowSerializationSchema serializationSchema =
- new AvroRowSerializationSchema(testData.f2.toString());
- final AvroRowDeserializationSchema deserializationSchema =
- new AvroRowDeserializationSchema(testData.f2.toString());
-
- if (legacyTimestampMapping) {
- final byte[] bytes = serializationSchema.serialize(testData.f1);
- deserializationSchema.deserialize(bytes);
- deserializationSchema.deserialize(bytes);
- final Row actual = deserializationSchema.deserialize(bytes);
- assertThat(actual).isEqualTo(testData.f1);
- } else {
- final byte[] bytes = serializationSchema.serialize(testData.f1, false);
- deserializationSchema.deserialize(bytes, false);
- deserializationSchema.deserialize(bytes, false);
- final Row actual = deserializationSchema.deserialize(bytes, false);
- assertThat(actual).isEqualTo(testData.f1);
- }
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- void testSerializability(boolean legacyTimestampMapping) throws Exception {
- final Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> testData =
- AvroTestUtils.getSpecificTestData();
- final String schemaString = testData.f1.getSchema().toString();
-
- // from class
- final AvroRowSerializationSchema classSer = new AvroRowSerializationSchema(testData.f0);
- final AvroRowDeserializationSchema classDeser =
- new AvroRowDeserializationSchema(testData.f0);
- testSerializability(classSer, classDeser, testData.f2, legacyTimestampMapping);
-
- // from schema string
- final AvroRowSerializationSchema schemaSer = new AvroRowSerializationSchema(schemaString);
- final AvroRowDeserializationSchema schemaDeser =
- new AvroRowDeserializationSchema(schemaString);
- testSerializability(schemaSer, schemaDeser, testData.f2, legacyTimestampMapping);
- }
-
- void testTimestampSerializeDeserializeLegacyMapping() throws Exception {
- final Tuple4<Class<? extends SpecificRecord>, SpecificRecord, GenericRecord, Row> testData =
- AvroTestUtils.getTimestampTestData();
-
- final String schemaString = testData.f1.getSchema().toString();
-
- final AvroRowSerializationSchema serializationSchema =
- new AvroRowSerializationSchema(schemaString);
- final AvroRowDeserializationSchema deserializationSchema =
- new AvroRowDeserializationSchema(schemaString);
-
- assertThatThrownBy(
- () -> {
- serializationSchema.serialize(testData.f3);
- })
- .isInstanceOf(RuntimeException.class)
- .hasMessage("Unsupported local timestamp type.");
-
- final byte[] bytes = serializationSchema.serialize(testData.f3, false);
-
- assertThatThrownBy(
- () -> {
- deserializationSchema.deserialize(bytes);
- })
- .isInstanceOf(RuntimeException.class)
- .hasMessage("Unsupported local timestamp type.");
- }
-
- @Test
- void testTimestampSpecificSerializeDeserializeNewMapping() throws Exception {
- final Tuple4<Class<? extends SpecificRecord>, SpecificRecord, GenericRecord, Row> testData =
- AvroTestUtils.getTimestampTestData();
-
- final String schemaString = testData.f1.getSchema().toString();
-
- final AvroRowSerializationSchema serializationSchema =
- new AvroRowSerializationSchema(schemaString);
- final AvroRowDeserializationSchema deserializationSchema =
- new AvroRowDeserializationSchema(schemaString);
-
- final byte[] bytes = serializationSchema.serialize(testData.f3, false);
- final Row actual = deserializationSchema.deserialize(bytes, false);
- assertThat(actual).isEqualTo(testData.f3);
- }
-
- @Test
- void testTimestampGenericGenericSerializeDeserializeNewMapping() throws Exception {
- final Tuple4<Class<? extends SpecificRecord>, SpecificRecord, GenericRecord, Row> testData =
- AvroTestUtils.getTimestampTestData();
-
- final String schemaString = testData.f2.getSchema().toString();
-
- final AvroRowSerializationSchema serializationSchema =
- new AvroRowSerializationSchema(schemaString);
- final AvroRowDeserializationSchema deserializationSchema =
- new AvroRowDeserializationSchema(schemaString);
-
- final byte[] bytes = serializationSchema.serialize(testData.f3);
- final Row actual = deserializationSchema.deserialize(bytes);
- assertThat(actual).isEqualTo(testData.f3);
- }
-
- @Test
- void testTimestampClassSerializeDeserializeNewMapping() throws Exception {
- final Tuple4<Class<? extends SpecificRecord>, SpecificRecord, GenericRecord, Row> testData =
- AvroTestUtils.getTimestampTestData();
-
- final AvroRowSerializationSchema serializationSchema =
- new AvroRowSerializationSchema(testData.f0);
- final AvroRowDeserializationSchema deserializationSchema =
- new AvroRowDeserializationSchema(testData.f0);
-
- final byte[] bytes = serializationSchema.serialize(testData.f3);
- final Row actual = deserializationSchema.deserialize(bytes);
- assertThat(actual).isEqualTo(testData.f3);
- }
-
- private void testSerializability(
- AvroRowSerializationSchema ser,
- AvroRowDeserializationSchema deser,
- Row data,
- boolean legacyTimestampMapping)
- throws Exception {
- final byte[] serBytes = InstantiationUtil.serializeObject(ser);
- final byte[] deserBytes = InstantiationUtil.serializeObject(deser);
-
- final AvroRowSerializationSchema serCopy =
- InstantiationUtil.deserializeObject(
- serBytes, Thread.currentThread().getContextClassLoader());
- final AvroRowDeserializationSchema deserCopy =
- InstantiationUtil.deserializeObject(
- deserBytes, Thread.currentThread().getContextClassLoader());
-
- if (legacyTimestampMapping) {
- final byte[] bytes = serCopy.serialize(data);
- deserCopy.deserialize(bytes);
- deserCopy.deserialize(bytes);
- final Row actual = deserCopy.deserialize(bytes);
- assertThat(actual).isEqualTo(data);
- } else {
- final byte[] bytes = serCopy.serialize(data, false);
- deserCopy.deserialize(bytes, false);
- deserCopy.deserialize(bytes, false);
- final Row actual = deserCopy.deserialize(bytes, false);
- assertThat(actual).isEqualTo(data);
- }
- }
-}
diff --git a/flink-python/pom.xml b/flink-python/pom.xml
index 7afc822145e..697a6d53cc3 100644
--- a/flink-python/pom.xml
+++ b/flink-python/pom.xml
@@ -97,6 +97,12 @@ under the License.
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-avro</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
<!-- Beam dependencies -->
@@ -721,6 +727,10 @@ under the License.
<pattern>com.google.protobuf</pattern>
<shadedPattern>org.apache.flink.api.python.shaded.com.google.protobuf</shadedPattern>
</relocation>
+ <relocation>
+ <pattern>org.apache.avro</pattern>
+ <shadedPattern>org.apache.flink.avro.shaded.org.apache.avro</shadedPattern>
+ </relocation>
</relocations>
</configuration>
</execution>
diff --git a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowDeserializationSchema.java b/flink-python/src/main/java/org/apache/flink/formats/avro/AvroRowDeserializationSchema.java
similarity index 98%
rename from flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowDeserializationSchema.java
rename to flink-python/src/main/java/org/apache/flink/formats/avro/AvroRowDeserializationSchema.java
index 74429b102f2..852ee54f850 100644
--- a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowDeserializationSchema.java
+++ b/flink-python/src/main/java/org/apache/flink/formats/avro/AvroRowDeserializationSchema.java
@@ -17,7 +17,7 @@
package org.apache.flink.formats.avro;
-import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.serialization.AbstractDeserializationSchema;
import org.apache.flink.api.common.serialization.DeserializationSchema;
@@ -80,13 +80,8 @@ import java.util.TimeZone;
*
* <p>Note: Changes in this class need to be kept in sync with the corresponding runtime class
* {@link AvroRowSerializationSchema} and schema converter {@link AvroSchemaConverter}.
- *
- * @deprecated The format was developed for the Table API users and will not be maintained for
- * DataStream API users anymore. Either use Table API or switch to Data Stream, defining your
- * own {@link DeserializationSchema}.
*/
-@PublicEvolving
-@Deprecated
+@Internal
public class AvroRowDeserializationSchema extends AbstractDeserializationSchema<Row> {
/** Used for time conversions into SQL types. */
private static final TimeZone LOCAL_TZ = TimeZone.getDefault();
diff --git a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowSerializationSchema.java b/flink-python/src/main/java/org/apache/flink/formats/avro/AvroRowSerializationSchema.java
similarity index 98%
rename from flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowSerializationSchema.java
rename to flink-python/src/main/java/org/apache/flink/formats/avro/AvroRowSerializationSchema.java
index 1e80306538c..90503e81915 100644
--- a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowSerializationSchema.java
+++ b/flink-python/src/main/java/org/apache/flink/formats/avro/AvroRowSerializationSchema.java
@@ -18,6 +18,7 @@
package org.apache.flink.formats.avro;
+import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.formats.avro.typeutils.AvroSchemaConverter;
@@ -66,12 +67,8 @@ import java.util.TimeZone;
*
* <p>Note: Changes in this class need to be kept in sync with the corresponding runtime class
* {@link AvroRowDeserializationSchema} and schema converter {@link AvroSchemaConverter}.
- *
- * @deprecated The format was developed for the Table API users and will not be maintained for
- * DataStream API users anymore. Either use Table API or switch to Data Stream, defining your
- * own {@link SerializationSchema}.
*/
-@Deprecated
+@Internal
public class AvroRowSerializationSchema implements SerializationSchema<Row> {
/** Used for time conversions from SQL types. */