This is an automated email from the ASF dual-hosted git repository. chesnay pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit fb95798b1c301152b912c4b8ec4a737ea16d8641 Author: Chesnay Schepler <[email protected]> AuthorDate: Tue Jul 26 16:24:43 2022 +0200 [FLINK-28807] Honor schema lifecycle --- .../sink/KinesisFirehoseSinkElementConverter.java | 30 ++++++++++++ .../sink/KinesisStreamsSinkElementConverter.java | 31 +++++++++++++ .../table/RowElasticsearchEmitter.java | 22 +++++++++ .../elasticsearch/ElasticsearchSinkBase.java | 2 +- .../elasticsearch/ElasticsearchSinkFunction.java | 10 +++- .../table/RowElasticsearchSinkFunction.java | 5 +- .../elasticsearch/ElasticsearchSinkBaseTest.java | 2 + .../kafka/sink/DefaultKafkaSinkContext.java | 2 +- .../JSONKeyValueDeserializationSchemaTest.java | 5 ++ flink-formats/flink-csv/pom.xml | 6 +++ .../flink/formats/csv/CsvFormatFactoryTest.java | 22 +++++++-- .../formats/csv/CsvRowDataSerDeSchemaTest.java | 3 ++ .../csv/CsvRowDeSerializationSchemaTest.java | 3 ++ .../json/canal/CanalJsonDeserializationSchema.java | 5 ++ .../json/canal/CanalJsonSerializationSchema.java | 3 +- .../DebeziumJsonDeserializationSchema.java | 5 ++ .../debezium/DebeziumJsonSerializationSchema.java | 3 +- .../maxwell/MaxwellJsonDeserializationSchema.java | 5 ++ .../maxwell/MaxwellJsonSerializationSchema.java | 1 + .../json/ogg/OggJsonDeserializationSchema.java | 5 ++ .../json/ogg/OggJsonSerializationSchema.java | 3 +- .../formats/json/JsonRowDataSerDeSchemaTest.java | 27 +++++++++++ .../json/JsonRowSerializationSchemaTest.java | 6 +++ .../json/canal/CanalJsonSerDeSchemaTest.java | 7 ++- .../json/debezium/DebeziumJsonSerDeSchemaTest.java | 5 +- .../json/maxwell/MaxwellJsonSerDerTest.java | 5 +- .../formats/json/ogg/OggJsonSerDeSchemaTest.java | 6 ++- .../utils/DeserializationSchemaMatcher.java | 2 + .../formats/utils/SerializationSchemaMatcher.java | 3 ++ flink-python/pom.xml | 14 ++++++ .../common/tests/test_serialization_schemas.py | 12 ++++- .../datastream/connectors/tests/test_kafka.py | 10 ++++ flink-python/pyflink/pyflink_gateway_server.py | 2 + .../testutils/formats/SchemaTestUtils.java | 53 ++++++++++++++++++++++ .../connector/upserttest/sink/UpsertTestSink.java | 8 ++++ 35 files changed, 316 insertions(+), 17 deletions(-) diff --git a/flink-connectors/flink-connector-aws-kinesis-firehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkElementConverter.java b/flink-connectors/flink-connector-aws-kinesis-firehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkElementConverter.java index dc725842ab0..b90db3387b9 100644 --- a/flink-connectors/flink-connector-aws-kinesis-firehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkElementConverter.java +++ b/flink-connectors/flink-connector-aws-kinesis-firehose/src/main/java/org/apache/flink/connector/firehose/sink/KinesisFirehoseSinkElementConverter.java @@ -21,7 +21,12 @@ import org.apache.flink.annotation.Internal; import org.apache.flink.api.common.serialization.SerializationSchema; import org.apache.flink.api.connector.sink2.SinkWriter; import org.apache.flink.connector.base.sink.writer.ElementConverter; +import org.apache.flink.metrics.MetricGroup; +import org.apache.flink.metrics.groups.UnregisteredMetricsGroup; +import org.apache.flink.util.FlinkRuntimeException; import org.apache.flink.util.Preconditions; +import org.apache.flink.util.SimpleUserCodeClassLoader; +import org.apache.flink.util.UserCodeClassLoader; import software.amazon.awssdk.core.SdkBytes; import software.amazon.awssdk.services.firehose.model.Record; @@ -34,6 +39,7 @@ import software.amazon.awssdk.services.firehose.model.Record; @Internal public class KinesisFirehoseSinkElementConverter<InputT> implements ElementConverter<InputT, Record> { + private boolean schemaOpened = false; /** A serialization schema to specify how the input element should be serialized. */ private final SerializationSchema<InputT> serializationSchema; @@ -44,11 +50,35 @@ public class KinesisFirehoseSinkElementConverter<InputT> @Override public Record apply(InputT element, SinkWriter.Context context) { + checkOpened(); return Record.builder() .data(SdkBytes.fromByteArray(serializationSchema.serialize(element))) .build(); } + private void checkOpened() { + if (!schemaOpened) { + try { + serializationSchema.open( + new SerializationSchema.InitializationContext() { + @Override + public MetricGroup getMetricGroup() { + return new UnregisteredMetricsGroup(); + } + + @Override + public UserCodeClassLoader getUserCodeClassLoader() { + return SimpleUserCodeClassLoader.create( + KinesisFirehoseSinkElementConverter.class.getClassLoader()); + } + }); + schemaOpened = true; + } catch (Exception e) { + throw new FlinkRuntimeException("Failed to initialize serialization schema.", e); + } + } + } + public static <InputT> Builder<InputT> builder() { return new Builder<>(); } diff --git a/flink-connectors/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/sink/KinesisStreamsSinkElementConverter.java b/flink-connectors/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/sink/KinesisStreamsSinkElementConverter.java index bca74421ee5..a7e441128f2 100644 --- a/flink-connectors/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/sink/KinesisStreamsSinkElementConverter.java +++ b/flink-connectors/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/sink/KinesisStreamsSinkElementConverter.java @@ -21,7 +21,12 @@ import org.apache.flink.annotation.Internal; import org.apache.flink.api.common.serialization.SerializationSchema; import org.apache.flink.api.connector.sink2.SinkWriter; import org.apache.flink.connector.base.sink.writer.ElementConverter; +import org.apache.flink.metrics.MetricGroup; +import org.apache.flink.metrics.groups.UnregisteredMetricsGroup; +import org.apache.flink.util.FlinkRuntimeException; import org.apache.flink.util.Preconditions; +import org.apache.flink.util.SimpleUserCodeClassLoader; +import org.apache.flink.util.UserCodeClassLoader; import software.amazon.awssdk.core.SdkBytes; import software.amazon.awssdk.services.kinesis.model.PutRecordsRequestEntry; @@ -43,6 +48,8 @@ public class KinesisStreamsSinkElementConverter<InputT> */ private final PartitionKeyGenerator<InputT> partitionKeyGenerator; + private boolean schemaOpened = false; + private KinesisStreamsSinkElementConverter( SerializationSchema<InputT> serializationSchema, PartitionKeyGenerator<InputT> partitionKeyGenerator) { @@ -52,12 +59,36 @@ public class KinesisStreamsSinkElementConverter<InputT> @Override public PutRecordsRequestEntry apply(InputT element, SinkWriter.Context context) { + checkOpened(); return PutRecordsRequestEntry.builder() .data(SdkBytes.fromByteArray(serializationSchema.serialize(element))) .partitionKey(partitionKeyGenerator.apply(element)) .build(); } + private void checkOpened() { + if (!schemaOpened) { + try { + serializationSchema.open( + new SerializationSchema.InitializationContext() { + @Override + public MetricGroup getMetricGroup() { + return new UnregisteredMetricsGroup(); + } + + @Override + public UserCodeClassLoader getUserCodeClassLoader() { + return SimpleUserCodeClassLoader.create( + KinesisStreamsSinkElementConverter.class.getClassLoader()); + } + }); + schemaOpened = true; + } catch (Exception e) { + throw new FlinkRuntimeException("Failed to initialize serialization schema.", e); + } + } + } + public static <InputT> Builder<InputT> builder() { return new Builder<>(); } diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/connector/elasticsearch/table/RowElasticsearchEmitter.java b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/connector/elasticsearch/table/RowElasticsearchEmitter.java index e90e0bca288..bddc6cb1c83 100644 --- a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/connector/elasticsearch/table/RowElasticsearchEmitter.java +++ b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/connector/elasticsearch/table/RowElasticsearchEmitter.java @@ -22,8 +22,13 @@ import org.apache.flink.api.common.serialization.SerializationSchema; import org.apache.flink.api.connector.sink2.SinkWriter; import org.apache.flink.connector.elasticsearch.sink.ElasticsearchEmitter; import org.apache.flink.connector.elasticsearch.sink.RequestIndexer; +import org.apache.flink.metrics.MetricGroup; +import org.apache.flink.metrics.groups.UnregisteredMetricsGroup; import org.apache.flink.table.api.TableException; import org.apache.flink.table.data.RowData; +import org.apache.flink.util.FlinkRuntimeException; +import org.apache.flink.util.SimpleUserCodeClassLoader; +import org.apache.flink.util.UserCodeClassLoader; import org.elasticsearch.action.ActionRequest; import org.elasticsearch.action.delete.DeleteRequest; @@ -61,6 +66,23 @@ class RowElasticsearchEmitter implements ElasticsearchEmitter<RowData> { @Override public void open() { + try { + serializationSchema.open( + new SerializationSchema.InitializationContext() { + @Override + public MetricGroup getMetricGroup() { + return new UnregisteredMetricsGroup(); + } + + @Override + public UserCodeClassLoader getUserCodeClassLoader() { + return SimpleUserCodeClassLoader.create( + RowElasticsearchEmitter.class.getClassLoader()); + } + }); + } catch (Exception e) { + throw new FlinkRuntimeException("Failed to initialize serialization schema.", e); + } indexGenerator.open(); } diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBase.java b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBase.java index 9ed83090a43..c0fad6f7c99 100644 --- a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBase.java +++ b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBase.java @@ -320,7 +320,7 @@ public abstract class ElasticsearchSinkBase<T, C extends AutoCloseable> extends callBridge.createBulkProcessorIndexer( bulkProcessor, flushOnCheckpoint, numPendingRequests); failureRequestIndexer = new BufferingNoOpRequestIndexer(); - elasticsearchSinkFunction.open(); + elasticsearchSinkFunction.open(getRuntimeContext()); } @Override diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkFunction.java b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkFunction.java index 0cf85239465..682d2f490e9 100644 --- a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkFunction.java +++ b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkFunction.java @@ -64,10 +64,18 @@ public interface ElasticsearchSinkFunction<T> extends Serializable, Function { /** * Initialization method for the function. It is called once before the actual working process - * methods. + * methods, if {@link #open(RuntimeContext)} is not overridden. */ default void open() throws Exception {} + /** + * Initialization method for the function. It is called once before the actual working process + * methods. + */ + default void open(RuntimeContext ctx) throws Exception { + open(); + } + /** Tear-down method for the function. It is called when the sink closes. */ default void close() throws Exception {} diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/RowElasticsearchSinkFunction.java b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/RowElasticsearchSinkFunction.java index 7fb1e4d4c15..48762522334 100644 --- a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/RowElasticsearchSinkFunction.java +++ b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/table/RowElasticsearchSinkFunction.java @@ -20,6 +20,7 @@ package org.apache.flink.streaming.connectors.elasticsearch.table; import org.apache.flink.annotation.Internal; import org.apache.flink.api.common.functions.RuntimeContext; +import org.apache.flink.api.common.serialization.RuntimeContextInitializationContextAdapters; import org.apache.flink.api.common.serialization.SerializationSchema; import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction; import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer; @@ -67,7 +68,9 @@ class RowElasticsearchSinkFunction implements ElasticsearchSinkFunction<RowData> } @Override - public void open() { + public void open(RuntimeContext ctx) throws Exception { + serializationSchema.open( + RuntimeContextInitializationContextAdapters.serializationAdapter(ctx)); indexGenerator.open(); } diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBaseTest.java b/flink-connectors/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBaseTest.java index 30417a52cc7..87480c520c1 100644 --- a/flink-connectors/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBaseTest.java +++ b/flink-connectors/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBaseTest.java @@ -24,6 +24,7 @@ import org.apache.flink.core.testutils.MultiShotLatch; import org.apache.flink.streaming.api.operators.StreamSink; import org.apache.flink.streaming.connectors.elasticsearch.util.NoOpFailureHandler; import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.apache.flink.streaming.util.MockStreamingRuntimeContext; import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness; import org.elasticsearch.action.ActionRequest; @@ -425,6 +426,7 @@ public class ElasticsearchSinkBaseTest { new DummyElasticsearchSink<>( new HashMap<>(), sinkFunction, new DummyRetryFailureHandler()); + sink.setRuntimeContext(new MockStreamingRuntimeContext(false, 1, 0)); sink.open(new Configuration()); sink.close(); diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/DefaultKafkaSinkContext.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/DefaultKafkaSinkContext.java index be4158d710b..bec5ab67e81 100644 --- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/DefaultKafkaSinkContext.java +++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/DefaultKafkaSinkContext.java @@ -32,7 +32,7 @@ import java.util.Properties; * Context providing information to assist constructing a {@link * org.apache.kafka.clients.producer.ProducerRecord}. */ -class DefaultKafkaSinkContext implements KafkaRecordSerializationSchema.KafkaSinkContext { +public class DefaultKafkaSinkContext implements KafkaRecordSerializationSchema.KafkaSinkContext { private final int subtaskId; private final int numberOfParallelInstances; diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/JSONKeyValueDeserializationSchemaTest.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/JSONKeyValueDeserializationSchemaTest.java index 0f67ba9c179..3a052cee2af 100644 --- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/JSONKeyValueDeserializationSchemaTest.java +++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/JSONKeyValueDeserializationSchemaTest.java @@ -17,6 +17,7 @@ package org.apache.flink.streaming.connectors.kafka; +import org.apache.flink.connector.testutils.formats.DummyInitializationContext; import org.apache.flink.streaming.util.serialization.JSONKeyValueDeserializationSchema; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; @@ -42,6 +43,7 @@ public class JSONKeyValueDeserializationSchemaTest { byte[] serializedValue = mapper.writeValueAsBytes(initialValue); JSONKeyValueDeserializationSchema schema = new JSONKeyValueDeserializationSchema(false); + schema.open(new DummyInitializationContext()); ObjectNode deserializedValue = schema.deserialize(newConsumerRecord(serializedKey, serializedValue)); @@ -60,6 +62,7 @@ public class JSONKeyValueDeserializationSchemaTest { byte[] serializedValue = mapper.writeValueAsBytes(initialValue); JSONKeyValueDeserializationSchema schema = new JSONKeyValueDeserializationSchema(false); + schema.open(new DummyInitializationContext()); ObjectNode deserializedValue = schema.deserialize(newConsumerRecord(serializedKey, serializedValue)); @@ -93,6 +96,7 @@ public class JSONKeyValueDeserializationSchemaTest { byte[] serializedValue = null; JSONKeyValueDeserializationSchema schema = new JSONKeyValueDeserializationSchema(false); + schema.open(new DummyInitializationContext()); ObjectNode deserializedValue = schema.deserialize(newConsumerRecord(serializedKey, serializedValue)); @@ -113,6 +117,7 @@ public class JSONKeyValueDeserializationSchemaTest { byte[] serializedValue = mapper.writeValueAsBytes(initialValue); JSONKeyValueDeserializationSchema schema = new JSONKeyValueDeserializationSchema(true); + schema.open(new DummyInitializationContext()); final ConsumerRecord<byte[], byte[]> consumerRecord = newConsumerRecord("topic#1", 3, 4L, serializedKey, serializedValue); ObjectNode deserializedValue = schema.deserialize(consumerRecord); diff --git a/flink-formats/flink-csv/pom.xml b/flink-formats/flink-csv/pom.xml index 0e3b0384a1f..daa54639d18 100644 --- a/flink-formats/flink-csv/pom.xml +++ b/flink-formats/flink-csv/pom.xml @@ -122,6 +122,12 @@ under the License. <type>test-jar</type> <scope>test</scope> </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-connector-test-utils</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> </dependencies> <build> diff --git a/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvFormatFactoryTest.java b/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvFormatFactoryTest.java index 468cc8ff959..22e84be6de7 100644 --- a/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvFormatFactoryTest.java +++ b/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvFormatFactoryTest.java @@ -48,6 +48,7 @@ import java.util.List; import java.util.Map; import java.util.function.Consumer; +import static org.apache.flink.connector.testutils.formats.SchemaTestUtils.open; import static org.apache.flink.core.testutils.FlinkMatchers.containsCause; import static org.apache.flink.table.data.DecimalData.fromBigDecimal; import static org.apache.flink.table.data.StringData.fromString; @@ -75,6 +76,7 @@ public class CsvFormatFactoryTest extends TestLogger { .setEscapeCharacter('\\') .setNullLiteral("n/a") .build(); + open(expectedDeser); final Map<String, String> options = getAllOptions(); DeserializationSchema<RowData> actualDeser = createDeserializationSchema(options); assertThat(actualDeser).isEqualTo(expectedDeser); @@ -215,6 +217,7 @@ public class CsvFormatFactoryTest extends TestLogger { DeserializationSchema<RowData> deserializationSchema = sourceMock.valueFormat.createRuntimeDecoder( ScanRuntimeProviderContext.INSTANCE, PHYSICAL_DATA_TYPE); + open(deserializationSchema); RowData expected = GenericRowData.of(fromString("abc"), 123, false); RowData actual = deserializationSchema.deserialize("abc\t123\tfalse".getBytes()); assertThat(actual).isEqualTo(expected); @@ -377,8 +380,11 @@ public class CsvFormatFactoryTest extends TestLogger { TestDynamicTableFactory.DynamicTableSourceMock sourceMock = createDynamicTableSourceMock(options); - return sourceMock.valueFormat.createRuntimeDecoder( - ScanRuntimeProviderContext.INSTANCE, PHYSICAL_DATA_TYPE); + final DeserializationSchema<RowData> schema = + sourceMock.valueFormat.createRuntimeDecoder( + ScanRuntimeProviderContext.INSTANCE, PHYSICAL_DATA_TYPE); + open(schema); + return schema; } private static DeserializationSchema<RowData> createDeserializationSchema( @@ -389,8 +395,11 @@ public class CsvFormatFactoryTest extends TestLogger { ProjectableDecodingFormat<DeserializationSchema<RowData>> valueFormat = (ProjectableDecodingFormat<DeserializationSchema<RowData>>) sourceMock.valueFormat; - return valueFormat.createRuntimeDecoder( - ScanRuntimeProviderContext.INSTANCE, PHYSICAL_DATA_TYPE, projections); + final DeserializationSchema<RowData> schema = + valueFormat.createRuntimeDecoder( + ScanRuntimeProviderContext.INSTANCE, PHYSICAL_DATA_TYPE, projections); + open(schema); + return schema; } private static TestDynamicTableFactory.DynamicTableSourceMock createDynamicTableSourceMock( @@ -407,6 +416,9 @@ public class CsvFormatFactoryTest extends TestLogger { TestDynamicTableFactory.DynamicTableSinkMock sinkMock = (TestDynamicTableFactory.DynamicTableSinkMock) actualSink; - return sinkMock.valueFormat.createRuntimeEncoder(null, PHYSICAL_DATA_TYPE); + final SerializationSchema<RowData> schema = + sinkMock.valueFormat.createRuntimeEncoder(null, PHYSICAL_DATA_TYPE); + open(schema); + return schema; } } diff --git a/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvRowDataSerDeSchemaTest.java b/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvRowDataSerDeSchemaTest.java index 31fd0ed0b07..6427c4b0d7a 100644 --- a/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvRowDataSerDeSchemaTest.java +++ b/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvRowDataSerDeSchemaTest.java @@ -39,6 +39,7 @@ import java.time.LocalDateTime; import java.time.LocalTime; import java.util.function.Consumer; +import static org.apache.flink.connector.testutils.formats.SchemaTestUtils.open; import static org.apache.flink.core.testutils.FlinkAssertions.anyCauseMatches; import static org.apache.flink.table.api.DataTypes.ARRAY; import static org.apache.flink.table.api.DataTypes.BIGINT; @@ -463,6 +464,7 @@ public class CsvRowDataSerDeSchemaTest { InstantiationUtil.deserializeObject( InstantiationUtil.serializeObject(serSchemaBuilder.build()), CsvRowDeSerializationSchemaTest.class.getClassLoader()); + open(schema); return schema.serialize(row); } @@ -475,6 +477,7 @@ public class CsvRowDataSerDeSchemaTest { InstantiationUtil.deserializeObject( InstantiationUtil.serializeObject(deserSchemaBuilder.build()), CsvRowDeSerializationSchemaTest.class.getClassLoader()); + open(schema); return schema.deserialize(csv != null ? csv.getBytes() : null); } diff --git a/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvRowDeSerializationSchemaTest.java b/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvRowDeSerializationSchemaTest.java index f23e7d6eba0..3f7b04f5723 100644 --- a/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvRowDeSerializationSchemaTest.java +++ b/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvRowDeSerializationSchemaTest.java @@ -35,6 +35,7 @@ import java.time.Instant; import java.time.LocalDateTime; import java.util.function.Consumer; +import static org.apache.flink.connector.testutils.formats.SchemaTestUtils.open; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -372,6 +373,7 @@ public class CsvRowDeSerializationSchemaTest { InstantiationUtil.deserializeObject( InstantiationUtil.serializeObject(serSchemaBuilder.build()), CsvRowDeSerializationSchemaTest.class.getClassLoader()); + open(schema); return schema.serialize(row); } @@ -383,6 +385,7 @@ public class CsvRowDeSerializationSchemaTest { InstantiationUtil.deserializeObject( InstantiationUtil.serializeObject(deserSchemaBuilder.build()), CsvRowDeSerializationSchemaTest.class.getClassLoader()); + open(schema); return schema.deserialize(csv.getBytes()); } } diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/canal/CanalJsonDeserializationSchema.java b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/canal/CanalJsonDeserializationSchema.java index d1bd3a37fe5..bb977783393 100644 --- a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/canal/CanalJsonDeserializationSchema.java +++ b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/canal/CanalJsonDeserializationSchema.java @@ -198,6 +198,11 @@ public final class CanalJsonDeserializationSchema implements DeserializationSche // ------------------------------------------------------------------------------------------ + @Override + public void open(InitializationContext context) throws Exception { + jsonDeserializer.open(context); + } + @Override public RowData deserialize(byte[] message) throws IOException { throw new RuntimeException( diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/canal/CanalJsonSerializationSchema.java b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/canal/CanalJsonSerializationSchema.java index 90302d17436..362b9df6e6a 100644 --- a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/canal/CanalJsonSerializationSchema.java +++ b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/canal/CanalJsonSerializationSchema.java @@ -70,7 +70,8 @@ public class CanalJsonSerializationSchema implements SerializationSchema<RowData } @Override - public void open(InitializationContext context) { + public void open(InitializationContext context) throws Exception { + jsonSerializer.open(context); reuse = new GenericRowData(2); } diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/debezium/DebeziumJsonDeserializationSchema.java b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/debezium/DebeziumJsonDeserializationSchema.java index d12b7cc3443..cba336d4f89 100644 --- a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/debezium/DebeziumJsonDeserializationSchema.java +++ b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/debezium/DebeziumJsonDeserializationSchema.java @@ -115,6 +115,11 @@ public final class DebeziumJsonDeserializationSchema implements DeserializationS this.ignoreParseErrors = ignoreParseErrors; } + @Override + public void open(InitializationContext context) throws Exception { + jsonDeserializer.open(context); + } + @Override public RowData deserialize(byte[] message) { throw new RuntimeException( diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/debezium/DebeziumJsonSerializationSchema.java b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/debezium/DebeziumJsonSerializationSchema.java index f6e37073dee..0dc9a96b012 100644 --- a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/debezium/DebeziumJsonSerializationSchema.java +++ b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/debezium/DebeziumJsonSerializationSchema.java @@ -67,7 +67,8 @@ public class DebeziumJsonSerializationSchema implements SerializationSchema<RowD } @Override - public void open(InitializationContext context) { + public void open(InitializationContext context) throws Exception { + jsonSerializer.open(context); genericRowData = new GenericRowData(3); } diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/maxwell/MaxwellJsonDeserializationSchema.java b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/maxwell/MaxwellJsonDeserializationSchema.java index 863620fd908..f58bfd4a1b4 100644 --- a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/maxwell/MaxwellJsonDeserializationSchema.java +++ b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/maxwell/MaxwellJsonDeserializationSchema.java @@ -109,6 +109,11 @@ public class MaxwellJsonDeserializationSchema implements DeserializationSchema<R this.fieldCount = physicalRowType.getFieldCount(); } + @Override + public void open(InitializationContext context) throws Exception { + jsonDeserializer.open(context); + } + @Override public RowData deserialize(byte[] message) throws IOException { throw new RuntimeException( diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/maxwell/MaxwellJsonSerializationSchema.java b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/maxwell/MaxwellJsonSerializationSchema.java index 8dfac669c5a..1fe567b08c3 100644 --- a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/maxwell/MaxwellJsonSerializationSchema.java +++ b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/maxwell/MaxwellJsonSerializationSchema.java @@ -69,6 +69,7 @@ public class MaxwellJsonSerializationSchema implements SerializationSchema<RowDa @Override public void open(InitializationContext context) throws Exception { + jsonSerializer.open(context); this.reuse = new GenericRowData(2); } diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/ogg/OggJsonDeserializationSchema.java b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/ogg/OggJsonDeserializationSchema.java index 4337beddaf3..85bd52dc02d 100644 --- a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/ogg/OggJsonDeserializationSchema.java +++ b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/ogg/OggJsonDeserializationSchema.java @@ -143,6 +143,11 @@ public final class OggJsonDeserializationSchema implements DeserializationSchema return jsonRowType.getFieldNames().indexOf(metadata.requiredJsonField.getName()); } + @Override + public void open(InitializationContext context) throws Exception { + jsonDeserializer.open(context); + } + @Override public RowData deserialize(byte[] message) { throw new RuntimeException( diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/ogg/OggJsonSerializationSchema.java b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/ogg/OggJsonSerializationSchema.java index 2189f86c3be..635ff3dc7e3 100644 --- a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/ogg/OggJsonSerializationSchema.java +++ b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/ogg/OggJsonSerializationSchema.java @@ -79,7 +79,8 @@ public class OggJsonSerializationSchema implements SerializationSchema<RowData> } @Override - public void open(InitializationContext context) { + public void open(InitializationContext context) throws Exception { + jsonSerializer.open(context); genericRowData = new GenericRowData(3); } diff --git a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowDataSerDeSchemaTest.java b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowDataSerDeSchemaTest.java index fb30702bf8b..888be3b9fe7 100644 --- a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowDataSerDeSchemaTest.java +++ b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowDataSerDeSchemaTest.java @@ -19,6 +19,7 @@ package org.apache.flink.formats.json; import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.connector.testutils.formats.DummyInitializationContext; import org.apache.flink.core.testutils.FlinkAssertions; import org.apache.flink.formats.common.TimestampFormat; import org.apache.flink.table.data.GenericMapData; @@ -52,6 +53,7 @@ import java.util.Map; import java.util.Random; import java.util.concurrent.ThreadLocalRandom; +import static org.apache.flink.connector.testutils.formats.SchemaTestUtils.open; import static org.apache.flink.core.testutils.FlinkAssertions.anyCauseMatches; import static org.apache.flink.table.api.DataTypes.ARRAY; import static org.apache.flink.table.api.DataTypes.BIGINT; @@ -165,6 +167,7 @@ class JsonRowDataSerDeSchemaTest { JsonRowDataDeserializationSchema deserializationSchema = new JsonRowDataDeserializationSchema( schema, resultTypeInfo, false, false, TimestampFormat.ISO_8601); + open(deserializationSchema); Row expected = new Row(18); expected.setField(0, true); @@ -198,6 +201,7 @@ class JsonRowDataSerDeSchemaTest { JsonFormatOptions.MapNullKeyMode.LITERAL, "null", true); + open(serializationSchema); byte[] actualBytes = serializationSchema.serialize(rowData); assertThat(serializedJson).containsExactly(actualBytes); @@ -246,6 +250,7 @@ class JsonRowDataSerDeSchemaTest { false, false, TimestampFormat.ISO_8601); + open(deserializationSchema); Row expected = new Row(7); expected.setField(0, bool); @@ -281,6 +286,7 @@ class JsonRowDataSerDeSchemaTest { false, false, TimestampFormat.ISO_8601); + open(deserializationSchema); JsonRowDataSerializationSchema serializationSchema = new JsonRowDataSerializationSchema( rowType, @@ -288,6 +294,7 @@ class JsonRowDataSerDeSchemaTest { JsonFormatOptions.MapNullKeyMode.LITERAL, "null", true); + open(serializationSchema); ObjectMapper objectMapper = new ObjectMapper(); @@ -366,6 +373,7 @@ class JsonRowDataSerDeSchemaTest { false, true, TimestampFormat.ISO_8601); + open(deserializationSchema); JsonRowDataSerializationSchema serializationSchema = new JsonRowDataSerializationSchema( rowType, @@ -373,6 +381,7 @@ class JsonRowDataSerDeSchemaTest { JsonFormatOptions.MapNullKeyMode.LITERAL, "null", true); + open(serializationSchema); for (int i = 0; i < jsons.length; i++) { String json = jsons[i]; @@ -390,6 +399,7 @@ class JsonRowDataSerDeSchemaTest { JsonRowDataDeserializationSchema deserializationSchema = new JsonRowDataDeserializationSchema( schema, InternalTypeInfo.of(schema), true, false, TimestampFormat.ISO_8601); + open(deserializationSchema); assertThat(deserializationSchema.deserialize(null)).isNull(); } @@ -402,6 +412,7 @@ class JsonRowDataSerDeSchemaTest { JsonRowDataDeserializationSchema deserializationSchema = new JsonRowDataDeserializationSchema( schema, InternalTypeInfo.of(schema), true, false, TimestampFormat.ISO_8601); + open(deserializationSchema); RowData rowData = deserializationSchema.deserialize("".getBytes()); assertThat(rowData).isNull(); } @@ -426,6 +437,7 @@ class JsonRowDataSerDeSchemaTest { false, false, TimestampFormat.ISO_8601); + open(deserializationSchema); Row expected = new Row(1); Row actual = convertToExternal(deserializationSchema.deserialize(serializedJson), dataType); @@ -435,6 +447,7 @@ class JsonRowDataSerDeSchemaTest { deserializationSchema = new JsonRowDataDeserializationSchema( schema, InternalTypeInfo.of(schema), true, false, TimestampFormat.ISO_8601); + open(deserializationSchema); String errorMessage = "Failed to deserialize JSON '{\"id\":123123123}'."; @@ -446,6 +459,7 @@ class JsonRowDataSerDeSchemaTest { deserializationSchema = new JsonRowDataDeserializationSchema( schema, InternalTypeInfo.of(schema), false, true, TimestampFormat.ISO_8601); + open(deserializationSchema); actual = convertToExternal(deserializationSchema.deserialize(serializedJson), dataType); assertThat(actual).isEqualTo(expected); @@ -480,6 +494,7 @@ class JsonRowDataSerDeSchemaTest { JsonRowDataDeserializationSchema deserializationSchema = new JsonRowDataDeserializationSchema( rowType, InternalTypeInfo.of(rowType), false, false, TimestampFormat.SQL); + open(deserializationSchema); JsonRowDataSerializationSchema serializationSchema = new JsonRowDataSerializationSchema( rowType, @@ -487,6 +502,7 @@ class JsonRowDataSerDeSchemaTest { JsonFormatOptions.MapNullKeyMode.LITERAL, "null", true); + open(serializationSchema); ObjectMapper objectMapper = new ObjectMapper(); @@ -530,6 +546,7 @@ class JsonRowDataSerDeSchemaTest { JsonFormatOptions.MapNullKeyMode.FAIL, "null", true); + open(serializationSchema1); // expect message for serializationSchema1 String errorMessage1 = "JSON format doesn't support to serialize map data with null keys." @@ -542,6 +559,7 @@ class JsonRowDataSerDeSchemaTest { JsonFormatOptions.MapNullKeyMode.DROP, "null", true); + open(serializationSchema2); // expect result for serializationSchema2 String expectResult2 = "{\"nestedMap\":{\"no-null key\":{\"no-null key\":1}}}"; @@ -552,6 +570,7 @@ class JsonRowDataSerDeSchemaTest { JsonFormatOptions.MapNullKeyMode.LITERAL, "nullKey", true); + open(serializationSchema3); // expect result for serializationSchema3 String expectResult3 = "{\"nestedMap\":{\"no-null key\":{\"no-null key\":1,\"nullKey\":2},\"nullKey\":{\"no-null key\":1,\"nullKey\":2}}}"; @@ -583,6 +602,7 @@ class JsonRowDataSerDeSchemaTest { JsonRowDataDeserializationSchema deserializer = new JsonRowDataDeserializationSchema( schema, resultTypeInfo, false, false, TimestampFormat.ISO_8601); + deserializer.open(new DummyInitializationContext()); JsonRowDataSerializationSchema plainDecimalSerializer = new JsonRowDataSerializationSchema( @@ -591,6 +611,7 @@ class JsonRowDataSerDeSchemaTest { JsonFormatOptions.MapNullKeyMode.LITERAL, "null", true); + plainDecimalSerializer.open(new DummyInitializationContext()); JsonRowDataSerializationSchema scientificDecimalSerializer = new JsonRowDataSerializationSchema( schema, @@ -598,6 +619,7 @@ class JsonRowDataSerDeSchemaTest { JsonFormatOptions.MapNullKeyMode.LITERAL, "null", false); + scientificDecimalSerializer.open(new DummyInitializationContext()); String plainDecimalJson = "{\"decimal1\":123.456789,\"decimal2\":454621864049246170,\"decimal3\":0.000000027}"; @@ -636,6 +658,7 @@ class JsonRowDataSerDeSchemaTest { JsonFormatOptions.MapNullKeyMode.FAIL, "null", true); + open(serializationSchema); String errorMessage = "Fail to serialize at field: f1."; assertThatThrownBy(() -> serializationSchema.serialize(genericRowData)) @@ -649,6 +672,7 @@ class JsonRowDataSerDeSchemaTest { JsonRowDataDeserializationSchema deserializationSchema = new JsonRowDataDeserializationSchema( rowType, InternalTypeInfo.of(rowType), false, false, TimestampFormat.SQL); + open(deserializationSchema); String errorMessage = "Fail to deserialize at field: f1."; assertThatThrownBy(() -> deserializationSchema.deserialize(json.getBytes())) @@ -664,6 +688,8 @@ class JsonRowDataSerDeSchemaTest { false, true, spec.timestampFormat); + ignoreErrorsSchema.open(new DummyInitializationContext()); + Row expected; if (spec.expected != null) { expected = spec.expected; @@ -686,6 +712,7 @@ class JsonRowDataSerDeSchemaTest { false, false, spec.timestampFormat); + open(failingSchema); assertThatThrownBy(() -> failingSchema.deserialize(spec.json.getBytes())) .hasMessageContaining(spec.errorMessage); diff --git a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowSerializationSchemaTest.java b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowSerializationSchemaTest.java index e9c46be1eb0..ac1fe181867 100644 --- a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowSerializationSchemaTest.java +++ b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowSerializationSchemaTest.java @@ -30,6 +30,7 @@ import java.sql.Time; import java.sql.Timestamp; import java.util.concurrent.ThreadLocalRandom; +import static org.apache.flink.connector.testutils.formats.SchemaTestUtils.open; import static org.apache.flink.formats.utils.SerializationSchemaMatcher.whenSerializedWith; import static org.hamcrest.Matchers.instanceOf; import static org.junit.Assert.assertEquals; @@ -81,8 +82,10 @@ public class JsonRowSerializationSchemaTest { final JsonRowSerializationSchema serializationSchema = new JsonRowSerializationSchema.Builder(rowSchema).build(); + open(serializationSchema); final JsonRowDeserializationSchema deserializationSchema = new JsonRowDeserializationSchema.Builder(rowSchema).build(); + open(deserializationSchema); byte[] bytes = serializationSchema.serialize(row1); assertEquals(row1, deserializationSchema.deserialize(bytes)); @@ -122,8 +125,10 @@ public class JsonRowSerializationSchemaTest { Types.PRIMITIVE_ARRAY(Types.INT)); JsonRowDeserializationSchema deserializationSchema = new JsonRowDeserializationSchema.Builder(schema).build(); + open(deserializationSchema); JsonRowSerializationSchema serializationSchema = JsonRowSerializationSchema.builder().withTypeInfo(schema).build(); + open(serializationSchema); for (int i = 0; i < jsons.length; i++) { String json = jsons[i]; @@ -173,6 +178,7 @@ public class JsonRowSerializationSchemaTest { final JsonRowSerializationSchema serializationSchema = new JsonRowSerializationSchema.Builder(rowSchema).build(); + open(serializationSchema); assertThat( row, whenSerializedWith(serializationSchema) diff --git a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/canal/CanalJsonSerDeSchemaTest.java b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/canal/CanalJsonSerDeSchemaTest.java index 1083ae6564e..e45bfcc5eee 100644 --- a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/canal/CanalJsonSerDeSchemaTest.java +++ b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/canal/CanalJsonSerDeSchemaTest.java @@ -18,6 +18,7 @@ package org.apache.flink.formats.json.canal; +import org.apache.flink.connector.testutils.formats.DummyInitializationContext; import org.apache.flink.formats.common.TimestampFormat; import org.apache.flink.formats.json.JsonFormatOptions; import org.apache.flink.formats.json.canal.CanalJsonDecodingFormat.ReadableMetadata; @@ -44,6 +45,7 @@ import java.util.List; import java.util.function.Consumer; import java.util.stream.Collectors; +import static org.apache.flink.connector.testutils.formats.SchemaTestUtils.open; import static org.apache.flink.table.api.DataTypes.FIELD; import static org.apache.flink.table.api.DataTypes.FLOAT; import static org.apache.flink.table.api.DataTypes.INT; @@ -80,6 +82,7 @@ class CanalJsonSerDeSchemaTest { final List<ReadableMetadata> requestedMetadata = Arrays.asList(ReadableMetadata.values()); final CanalJsonDeserializationSchema deserializationSchema = createCanalJsonDeserializationSchema(null, null, requestedMetadata); + open(deserializationSchema); final SimpleCollector collector = new SimpleCollector(); deserializationSchema.deserialize(null, collector); @@ -139,6 +142,7 @@ class CanalJsonSerDeSchemaTest { public void runTest(List<String> lines, CanalJsonDeserializationSchema deserializationSchema) throws Exception { + open(deserializationSchema); SimpleCollector collector = new SimpleCollector(); for (String line : lines) { deserializationSchema.deserialize(line.getBytes(StandardCharsets.UTF_8), collector); @@ -215,7 +219,7 @@ class CanalJsonSerDeSchemaTest { JsonFormatOptions.MapNullKeyMode.LITERAL, "null", true); - serializationSchema.open(null); + serializationSchema.open(new DummyInitializationContext()); List<String> result = new ArrayList<>(); for (RowData rowData : collector.list) { @@ -262,6 +266,7 @@ class CanalJsonSerDeSchemaTest { final List<ReadableMetadata> requestedMetadata = Arrays.asList(ReadableMetadata.values()); final CanalJsonDeserializationSchema deserializationSchema = createCanalJsonDeserializationSchema(database, table, requestedMetadata); + open(deserializationSchema); final SimpleCollector collector = new SimpleCollector(); deserializationSchema.deserialize(firstLine.getBytes(StandardCharsets.UTF_8), collector); diff --git a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/debezium/DebeziumJsonSerDeSchemaTest.java b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/debezium/DebeziumJsonSerDeSchemaTest.java index ef3b1d8ab5a..3b9151f33a9 100644 --- a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/debezium/DebeziumJsonSerDeSchemaTest.java +++ b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/debezium/DebeziumJsonSerDeSchemaTest.java @@ -44,6 +44,7 @@ import java.util.List; import java.util.function.Consumer; import java.util.stream.Collectors; +import static org.apache.flink.connector.testutils.formats.SchemaTestUtils.open; import static org.apache.flink.table.api.DataTypes.FIELD; import static org.apache.flink.table.api.DataTypes.FLOAT; import static org.apache.flink.table.api.DataTypes.INT; @@ -179,6 +180,7 @@ class DebeziumJsonSerDeSchemaTest { schemaInclude, false, TimestampFormat.ISO_8601); + open(deserializationSchema); SimpleCollector collector = new SimpleCollector(); for (String line : lines) { @@ -249,7 +251,7 @@ class DebeziumJsonSerDeSchemaTest { "null", true); - serializationSchema.open(null); + open(serializationSchema); actual = new ArrayList<>(); for (RowData rowData : collector.list) { actual.add(new String(serializationSchema.serialize(rowData), StandardCharsets.UTF_8)); @@ -303,6 +305,7 @@ class DebeziumJsonSerDeSchemaTest { schemaInclude, false, TimestampFormat.ISO_8601); + open(deserializationSchema); final SimpleCollector collector = new SimpleCollector(); deserializationSchema.deserialize(firstLine.getBytes(StandardCharsets.UTF_8), collector); diff --git a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/maxwell/MaxwellJsonSerDerTest.java b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/maxwell/MaxwellJsonSerDerTest.java index ffd68b20be3..12d64fd99d0 100644 --- a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/maxwell/MaxwellJsonSerDerTest.java +++ b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/maxwell/MaxwellJsonSerDerTest.java @@ -43,6 +43,7 @@ import java.util.Collections; import java.util.List; import java.util.stream.Collectors; +import static org.apache.flink.connector.testutils.formats.SchemaTestUtils.open; import static org.apache.flink.table.api.DataTypes.FIELD; import static org.apache.flink.table.api.DataTypes.FLOAT; import static org.apache.flink.table.api.DataTypes.INT; @@ -80,6 +81,7 @@ class MaxwellJsonSerDerTest { InternalTypeInfo.of(producedDataType.getLogicalType()), false, TimestampFormat.ISO_8601); + open(deserializationSchema); final SimpleCollector collector = new SimpleCollector(); deserializationSchema.deserialize(firstLine.getBytes(StandardCharsets.UTF_8), collector); assertThat(collector.list).hasSize(1); @@ -109,6 +111,7 @@ class MaxwellJsonSerDerTest { InternalTypeInfo.of(PHYSICAL_DATA_TYPE.getLogicalType()), false, TimestampFormat.ISO_8601); + open(deserializationSchema); SimpleCollector collector = new SimpleCollector(); for (String line : lines) { @@ -185,7 +188,7 @@ class MaxwellJsonSerDerTest { JsonFormatOptions.MapNullKeyMode.LITERAL, "null", true); - serializationSchema.open(null); + open(serializationSchema); List<String> result = new ArrayList<>(); for (RowData rowData : collector.list) { result.add(new String(serializationSchema.serialize(rowData), StandardCharsets.UTF_8)); diff --git a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/ogg/OggJsonSerDeSchemaTest.java b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/ogg/OggJsonSerDeSchemaTest.java index 9b694609d22..2fa78c89412 100644 --- a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/ogg/OggJsonSerDeSchemaTest.java +++ b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/ogg/OggJsonSerDeSchemaTest.java @@ -44,6 +44,7 @@ import java.util.Collections; import java.util.List; import java.util.stream.Collectors; +import static org.apache.flink.connector.testutils.formats.SchemaTestUtils.open; import static org.apache.flink.table.api.DataTypes.FIELD; import static org.apache.flink.table.api.DataTypes.FLOAT; import static org.apache.flink.table.api.DataTypes.INT; @@ -87,6 +88,7 @@ class OggJsonSerDeSchemaTest { InternalTypeInfo.of(PHYSICAL_DATA_TYPE.getLogicalType()), false, TimestampFormat.ISO_8601); + open(deserializationSchema); SimpleCollector collector = new SimpleCollector(); deserializationSchema.deserialize(null, collector); deserializationSchema.deserialize(new byte[] {}, collector); @@ -112,6 +114,7 @@ class OggJsonSerDeSchemaTest { InternalTypeInfo.of(producedDataTypes.getLogicalType()), false, TimestampFormat.ISO_8601); + open(deserializationSchema); final SimpleCollector collector = new SimpleCollector(); deserializationSchema.deserialize(firstLine.getBytes(StandardCharsets.UTF_8), collector); @@ -144,6 +147,7 @@ class OggJsonSerDeSchemaTest { InternalTypeInfo.of(PHYSICAL_DATA_TYPE.getLogicalType()), false, TimestampFormat.ISO_8601); + open(deserializationSchema); SimpleCollector collector = new SimpleCollector(); for (String line : lines) { @@ -214,7 +218,7 @@ class OggJsonSerDeSchemaTest { "null", true); - serializationSchema.open(null); + open(serializationSchema); actual = new ArrayList<>(); for (RowData rowData : collector.list) { actual.add(new String(serializationSchema.serialize(rowData), StandardCharsets.UTF_8)); diff --git a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/utils/DeserializationSchemaMatcher.java b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/utils/DeserializationSchemaMatcher.java index 59ee178200f..c6d9e6229b0 100644 --- a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/utils/DeserializationSchemaMatcher.java +++ b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/utils/DeserializationSchemaMatcher.java @@ -28,6 +28,7 @@ import org.hamcrest.TypeSafeMatcher; import java.io.IOException; import java.util.Objects; +import static org.apache.flink.connector.testutils.formats.SchemaTestUtils.open; import static org.apache.flink.util.InstantiationUtil.deserializeObject; import static org.apache.flink.util.InstantiationUtil.serializeObject; @@ -141,6 +142,7 @@ public abstract class DeserializationSchemaMatcher extends TypeSafeMatcher<byte[ deserializeObject( serializeObject(deserializationSchema), this.getClass().getClassLoader()); + open(this.deserializationSchema); } catch (IOException | ClassNotFoundException e) { throw new RuntimeException(e); } diff --git a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/utils/SerializationSchemaMatcher.java b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/utils/SerializationSchemaMatcher.java index be095155106..1cf6f6d20c8 100644 --- a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/utils/SerializationSchemaMatcher.java +++ b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/utils/SerializationSchemaMatcher.java @@ -29,6 +29,7 @@ import org.hamcrest.TypeSafeMatcher; import java.io.IOException; import java.util.Objects; +import static org.apache.flink.connector.testutils.formats.SchemaTestUtils.open; import static org.apache.flink.util.InstantiationUtil.deserializeObject; import static org.apache.flink.util.InstantiationUtil.serializeObject; @@ -150,10 +151,12 @@ public abstract class SerializationSchemaMatcher extends TypeSafeMatcher<Row> { deserializeObject( serializeObject(serializationSchema), this.getClass().getClassLoader()); + open(this.serializationSchema); this.deserializationSchema = deserializeObject( serializeObject(deserializationSchema), this.getClass().getClassLoader()); + open(this.deserializationSchema); } catch (IOException | ClassNotFoundException e) { throw new RuntimeException(e); } diff --git a/flink-python/pom.xml b/flink-python/pom.xml index 6d5b18385b3..bc376f844a6 100644 --- a/flink-python/pom.xml +++ b/flink-python/pom.xml @@ -173,6 +173,20 @@ under the License. <scope>test</scope> </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-connector-test-utils</artifactId> + <version>${project.version}</version> + <scope>test</scope> + <exclusions> + <exclusion> + <!-- for dependency convergence --> + <groupId>org.apache.curator</groupId> + <artifactId>curator-test</artifactId> + </exclusion> + </exclusions> + </dependency> + <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-runtime</artifactId> diff --git a/flink-python/pyflink/common/tests/test_serialization_schemas.py b/flink-python/pyflink/common/tests/test_serialization_schemas.py index ba3f6857953..9544bd13dba 100644 --- a/flink-python/pyflink/common/tests/test_serialization_schemas.py +++ b/flink-python/pyflink/common/tests/test_serialization_schemas.py @@ -35,6 +35,7 @@ class TestRowSerializationSchemas(PyFlinkTestCase): .deserialize(expected_string.encode(encoding='utf-8'))) def test_json_row_serialization_deserialization_schema(self): + jvm = get_gateway().jvm jsons = ["{\"svt\":\"2020-02-24T12:58:09.209+0800\"}", "{\"svt\":\"2020-02-24T12:58:09.209+0800\", " "\"ops\":{\"id\":\"281708d0-4092-4c21-9233-931950b6eccf\"},\"ids\":[1, 2, 3]}", @@ -54,6 +55,10 @@ class TestRowSerializationSchemas(PyFlinkTestCase): .with_type_info(row_schema).build() json_row_deserialization_schema = JsonRowDeserializationSchema.builder() \ .type_info(row_schema).build() + json_row_serialization_schema._j_serialization_schema.open( + jvm.org.apache.flink.connector.testutils.formats.DummyInitializationContext()) + json_row_deserialization_schema._j_deserialization_schema.open( + jvm.org.apache.flink.connector.testutils.formats.DummyInitializationContext()) for i in range(len(jsons)): j_row = json_row_deserialization_schema._j_deserialization_schema\ @@ -63,7 +68,8 @@ class TestRowSerializationSchemas(PyFlinkTestCase): self.assertEqual(expected_jsons[i], result) def test_csv_row_serialization_schema(self): - JRow = get_gateway().jvm.org.apache.flink.types.Row + jvm = get_gateway().jvm + JRow = jvm.org.apache.flink.types.Row j_row = JRow(3) j_row.setField(0, "BEGIN") @@ -80,6 +86,10 @@ class TestRowSerializationSchemas(PyFlinkTestCase): csv_row_deserialization_schema = CsvRowDeserializationSchema.Builder(row_info)\ .set_escape_character('*').set_quote_character('\'')\ .set_array_element_delimiter(':').set_field_delimiter(';').build() + csv_row_serialization_schema._j_serialization_schema.open( + jvm.org.apache.flink.connector.testutils.formats.DummyInitializationContext()) + csv_row_deserialization_schema._j_deserialization_schema.open( + jvm.org.apache.flink.connector.testutils.formats.DummyInitializationContext()) serialized_bytes = csv_row_serialization_schema._j_serialization_schema.serialize(j_row) self.assertEqual(expected_csv, str(serialized_bytes, encoding='utf-8')) diff --git a/flink-python/pyflink/datastream/connectors/tests/test_kafka.py b/flink-python/pyflink/datastream/connectors/tests/test_kafka.py index a0b7761d6a9..de02b041333 100644 --- a/flink-python/pyflink/datastream/connectors/tests/test_kafka.py +++ b/flink-python/pyflink/datastream/connectors/tests/test_kafka.py @@ -464,6 +464,11 @@ class KafkaRecordSerializationSchemaTests(PyFlinkTestCase): .set_value_serialization_schema( JsonRowSerializationSchema.builder().with_type_info(input_type).build()) \ .build() + jvm = get_gateway().jvm + serialization_schema._j_serialization_schema.open( + jvm.org.apache.flink.connector.testutils.formats.DummyInitializationContext(), + jvm.org.apache.flink.connector.kafka.sink.DefaultKafkaSinkContext( + 0, 1, jvm.java.util.Properties())) j_record = serialization_schema._j_serialization_schema.serialize( to_java_data_structure(Row('test')), None, None @@ -490,6 +495,11 @@ class KafkaRecordSerializationSchemaTests(PyFlinkTestCase): .set_value_serialization_schema( JsonRowSerializationSchema.builder().with_type_info(input_type).build()) \ .build() + jvm = get_gateway().jvm + serialization_schema._j_serialization_schema.open( + jvm.org.apache.flink.connector.testutils.formats.DummyInitializationContext(), + jvm.org.apache.flink.connector.kafka.sink.DefaultKafkaSinkContext( + 0, 1, jvm.java.util.Properties())) sink = KafkaSink.builder() \ .set_bootstrap_servers('localhost:9092') \ .set_record_serializer(serialization_schema) \ diff --git a/flink-python/pyflink/pyflink_gateway_server.py b/flink-python/pyflink/pyflink_gateway_server.py index eabaa25566d..5231e9d1632 100644 --- a/flink-python/pyflink/pyflink_gateway_server.py +++ b/flink-python/pyflink/pyflink_gateway_server.py @@ -230,6 +230,8 @@ def construct_test_classpath(): "flink-connectors/flink-connector-cassandra/target/flink-connector-*.jar", "flink-python/target/artifacts/testDataStream.jar", "flink-python/target/flink-python*-tests.jar", + ("flink-test-utils-parent/flink-connector-test-utils/target/" + "flink-connector-test-utils-*.jar"), ("flink-state-backends/flink-statebackend-rocksdb/target/" "flink-statebackend-rocksdb*tests.jar"), ] diff --git a/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/formats/SchemaTestUtils.java b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/formats/SchemaTestUtils.java new file mode 100644 index 00000000000..34245802778 --- /dev/null +++ b/flink-test-utils-parent/flink-connector-test-utils/src/main/java/org/apache/flink/connector/testutils/formats/SchemaTestUtils.java @@ -0,0 +1,53 @@ +/* + * 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.connector.testutils.formats; + +import org.apache.flink.api.common.serialization.DeserializationSchema; +import org.apache.flink.api.common.serialization.SerializationSchema; + +/** Test utilities for schemas. */ +public class SchemaTestUtils { + + /** + * Opens the given schema with a mock initialization context. + * + * @param schema to open + * @throws RuntimeException if the schema throws an exception + */ + public static void open(SerializationSchema<?> schema) { + try { + schema.open(new DummyInitializationContext()); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + /** + * Opens the given schema with a mock initialization context. + * + * @param schema to open + * @throws RuntimeException if the schema throws an exception + */ + public static void open(DeserializationSchema<?> schema) { + try { + schema.open(new DummyInitializationContext()); + } catch (Exception e) { + throw new RuntimeException(e); + } + } +} diff --git a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/sink/UpsertTestSink.java b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/sink/UpsertTestSink.java index 27b57187a7a..b628ccbc55a 100644 --- a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/sink/UpsertTestSink.java +++ b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/sink/UpsertTestSink.java @@ -23,6 +23,7 @@ import org.apache.flink.annotation.PublicEvolving; import org.apache.flink.api.common.serialization.SerializationSchema; import org.apache.flink.api.connector.sink2.Sink; import org.apache.flink.api.connector.sink2.SinkWriter; +import org.apache.flink.util.FlinkRuntimeException; import java.io.File; @@ -67,6 +68,13 @@ public class UpsertTestSink<IN> implements Sink<IN> { @Internal @Override public SinkWriter<IN> createWriter(InitContext context) { + try { + keySerializationSchema.open(context.asSerializationSchemaInitializationContext()); + valueSerializationSchema.open(context.asSerializationSchemaInitializationContext()); + } catch (Exception e) { + throw new FlinkRuntimeException("Failed to initialize schema.", e); + } + return new UpsertTestSinkWriter<>( outputFile, keySerializationSchema, valueSerializationSchema); }
