specificStartupOffsets) {
+
+ return new Kafka09TableSource(
+ schema,
+ proctimeAttribute,
+ rowtimeAttributeDescriptors,
+ fieldMapping,
+ topic,
+ properties,
+ deserializationSchema,
+ startupMode,
+ specificStartupOffsets
+ );
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaAvroTableSource.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaAvroTableSource.java
index 8c8ce324ba997..3e9f2b03e344f 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaAvroTableSource.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaAvroTableSource.java
@@ -21,16 +21,14 @@
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.formats.avro.AvroRowDeserializationSchema;
-import org.apache.flink.formats.avro.typeutils.AvroSchemaConverter;
import org.apache.flink.table.api.TableSchema;
-import org.apache.flink.table.sources.DefinedFieldMapping;
+import org.apache.flink.table.descriptors.ConnectorDescriptor;
import org.apache.flink.table.sources.StreamTableSource;
import org.apache.avro.specific.SpecificRecord;
import org.apache.avro.specific.SpecificRecordBase;
import java.util.Map;
-import java.util.Objects;
import java.util.Properties;
/**
@@ -38,13 +36,15 @@
*
* The version-specific Kafka consumers need to extend this class and
* override {@link #createKafkaConsumer(String, Properties, DeserializationSchema)}}.
+ *
+ * @deprecated Use the {@link org.apache.flink.table.descriptors.Kafka} descriptor together
+ * with descriptors for schema and format instead. Descriptors allow for
+ * implementation-agnostic definition of tables. See also
+ * {@link org.apache.flink.table.api.TableEnvironment#from(ConnectorDescriptor)}.
*/
+@Deprecated
@Internal
-public abstract class KafkaAvroTableSource extends KafkaTableSource implements DefinedFieldMapping {
-
- private final Class avroRecordClass;
-
- private Map fieldMapping;
+public abstract class KafkaAvroTableSource extends KafkaTableSource {
/**
* Creates a generic Kafka Avro {@link StreamTableSource} using a given {@link SpecificRecord}.
@@ -53,7 +53,12 @@ public abstract class KafkaAvroTableSource extends KafkaTableSource implements D
* @param properties Properties for the Kafka consumer.
* @param schema Schema of the produced table.
* @param avroRecordClass Class of the Avro record that is read from the Kafka topic.
+ * @deprecated Use the {@link org.apache.flink.table.descriptors.Kafka} descriptor together
+ * with descriptors for schema and format instead. Descriptors allow for
+ * implementation-agnostic definition of tables. See also
+ * {@link org.apache.flink.table.api.TableEnvironment#from(ConnectorDescriptor)}.
*/
+ @Deprecated
protected KafkaAvroTableSource(
String topic,
Properties properties,
@@ -61,59 +66,15 @@ protected KafkaAvroTableSource(
Class avroRecordClass) {
super(
+ schema,
topic,
properties,
- schema,
- AvroSchemaConverter.convertToTypeInfo(avroRecordClass));
-
- this.avroRecordClass = avroRecordClass;
- }
-
- @Override
- public Map getFieldMapping() {
- return fieldMapping;
+ new AvroRowDeserializationSchema(avroRecordClass));
}
@Override
public String explainSource() {
- return "KafkaAvroTableSource(" + this.avroRecordClass.getSimpleName() + ")";
- }
-
- @Override
- protected AvroRowDeserializationSchema getDeserializationSchema() {
- return new AvroRowDeserializationSchema(avroRecordClass);
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (!(o instanceof KafkaAvroTableSource)) {
- return false;
- }
- if (!super.equals(o)) {
- return false;
- }
- final KafkaAvroTableSource that = (KafkaAvroTableSource) o;
- return Objects.equals(avroRecordClass, that.avroRecordClass) &&
- Objects.equals(fieldMapping, that.fieldMapping);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(super.hashCode(), avroRecordClass, fieldMapping);
- }
-
- //////// SETTERS FOR OPTIONAL PARAMETERS
-
- /**
- * Configures a field mapping for this TableSource.
- *
- * @param fieldMapping The field mapping.
- */
- protected void setFieldMapping(Map fieldMapping) {
- this.fieldMapping = fieldMapping;
+ return "KafkaAvroTableSource";
}
//////// HELPER METHODS
@@ -124,7 +85,13 @@ protected void setFieldMapping(Map fieldMapping) {
*
* @param Type of the KafkaAvroTableSource produced by the builder.
* @param Type of the KafkaAvroTableSource.Builder subclass.
+ *
+ * @deprecated Use the {@link org.apache.flink.table.descriptors.Kafka} descriptor together
+ * with descriptors for schema and format instead. Descriptors allow for
+ * implementation-agnostic definition of tables. See also
+ * {@link org.apache.flink.table.api.TableEnvironment#from(ConnectorDescriptor)}.
*/
+ @Deprecated
protected abstract static class Builder
extends KafkaTableSource.Builder {
@@ -137,7 +104,9 @@ protected abstract static class Builder avroClass) {
this.avroClass = avroClass;
return builder();
@@ -153,7 +122,9 @@ public B forAvroRecordClass(Class avroClass) {
*
* @param schemaToAvroMapping A mapping from schema fields to Avro fields.
* @return The builder.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
public B withTableToAvroMapping(Map schemaToAvroMapping) {
this.fieldMapping = schemaToAvroMapping;
return builder();
@@ -163,7 +134,9 @@ public B withTableToAvroMapping(Map schemaToAvroMapping) {
* Returns the configured Avro class.
*
* @return The configured Avro class.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
protected Class getAvroRecordClass() {
return this.avroClass;
}
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaAvroTableSourceFactory.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaAvroTableSourceFactory.java
deleted file mode 100644
index 8ef7270ccae5a..0000000000000
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaAvroTableSourceFactory.java
+++ /dev/null
@@ -1,81 +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.streaming.connectors.kafka;
-
-import org.apache.flink.formats.avro.typeutils.AvroSchemaConverter;
-import org.apache.flink.table.api.TableSchema;
-import org.apache.flink.table.descriptors.AvroValidator;
-import org.apache.flink.table.descriptors.DescriptorProperties;
-import org.apache.flink.table.descriptors.FormatDescriptorValidator;
-import org.apache.flink.table.descriptors.SchemaValidator;
-
-import org.apache.avro.specific.SpecificRecordBase;
-
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-
-/**
- * Factory for creating configured instances of {@link KafkaAvroTableSource}.
- */
-public abstract class KafkaAvroTableSourceFactory extends KafkaTableSourceFactory {
-
- @Override
- protected String formatType() {
- return AvroValidator.FORMAT_TYPE_VALUE;
- }
-
- @Override
- protected int formatPropertyVersion() {
- return 1;
- }
-
- @Override
- protected List formatProperties() {
- return Collections.singletonList(AvroValidator.FORMAT_RECORD_CLASS);
- }
-
- @Override
- protected FormatDescriptorValidator formatValidator() {
- return new AvroValidator();
- }
-
- @Override
- protected KafkaTableSource.Builder createBuilderWithFormat(DescriptorProperties params) {
- KafkaAvroTableSource.Builder builder = createKafkaAvroBuilder();
-
- // Avro format schema
- final Class avroRecordClass =
- params.getClass(AvroValidator.FORMAT_RECORD_CLASS, SpecificRecordBase.class);
- builder.forAvroRecordClass(avroRecordClass);
- final TableSchema formatSchema = TableSchema.fromTypeInfo(AvroSchemaConverter.convertToTypeInfo(avroRecordClass));
-
- // field mapping
- final Map mapping = SchemaValidator.deriveFieldMapping(params, Optional.of(formatSchema));
- builder.withTableToAvroMapping(mapping);
-
- return builder;
- }
-
- /**
- * Creates a version specific {@link KafkaAvroTableSource.Builder}.
- */
- protected abstract KafkaAvroTableSource.Builder createKafkaAvroBuilder();
-}
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSource.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSource.java
index b2bb8ff773bf4..bd0d0dedf22e7 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSource.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSource.java
@@ -20,14 +20,12 @@
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.serialization.DeserializationSchema;
-import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.formats.json.JsonRowDeserializationSchema;
import org.apache.flink.table.api.TableSchema;
-import org.apache.flink.table.sources.DefinedFieldMapping;
+import org.apache.flink.table.descriptors.ConnectorDescriptor;
import org.apache.flink.table.sources.StreamTableSource;
import java.util.Map;
-import java.util.Objects;
import java.util.Properties;
/**
@@ -37,15 +35,15 @@
* override {@link #createKafkaConsumer(String, Properties, DeserializationSchema)}}.
*
* The field names are used to parse the JSON file and so are the types.
+ *
+ * @deprecated Use the {@link org.apache.flink.table.descriptors.Kafka} descriptor together
+ * with descriptors for schema and format instead. Descriptors allow for
+ * implementation-agnostic definition of tables. See also
+ * {@link org.apache.flink.table.api.TableEnvironment#from(ConnectorDescriptor)}.
*/
+@Deprecated
@Internal
-public abstract class KafkaJsonTableSource extends KafkaTableSource implements DefinedFieldMapping {
-
- private TableSchema jsonSchema;
-
- private Map fieldMapping;
-
- private boolean failOnMissingField;
+public abstract class KafkaJsonTableSource extends KafkaTableSource {
/**
* Creates a generic Kafka JSON {@link StreamTableSource}.
@@ -54,7 +52,9 @@ public abstract class KafkaJsonTableSource extends KafkaTableSource implements D
* @param properties Properties for the Kafka consumer.
* @param tableSchema The schema of the table.
* @param jsonSchema The schema of the JSON messages to decode from Kafka.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
protected KafkaJsonTableSource(
String topic,
Properties properties,
@@ -62,24 +62,10 @@ protected KafkaJsonTableSource(
TableSchema jsonSchema) {
super(
+ tableSchema,
topic,
properties,
- tableSchema,
- jsonSchemaToReturnType(jsonSchema));
-
- this.jsonSchema = jsonSchema;
- }
-
- @Override
- public Map getFieldMapping() {
- return fieldMapping;
- }
-
- @Override
- protected JsonRowDeserializationSchema getDeserializationSchema() {
- JsonRowDeserializationSchema deserSchema = new JsonRowDeserializationSchema(jsonSchemaToReturnType(jsonSchema));
- deserSchema.setFailOnMissingField(failOnMissingField);
- return deserSchema;
+ new JsonRowDeserializationSchema(jsonSchema.toRowType()));
}
@Override
@@ -87,28 +73,6 @@ public String explainSource() {
return "KafkaJsonTableSource";
}
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (!(o instanceof KafkaJsonTableSource)) {
- return false;
- }
- if (!super.equals(o)) {
- return false;
- }
- KafkaJsonTableSource that = (KafkaJsonTableSource) o;
- return failOnMissingField == that.failOnMissingField &&
- Objects.equals(jsonSchema, that.jsonSchema) &&
- Objects.equals(fieldMapping, that.fieldMapping);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(super.hashCode(), jsonSchema, fieldMapping, failOnMissingField);
- }
-
//////// SETTERS FOR OPTIONAL PARAMETERS
/**
@@ -116,34 +80,27 @@ public int hashCode() {
* TableSource will fail for missing fields if set to true. If set to false, the missing field is set to null.
*
* @param failOnMissingField Flag that specifies the TableSource behavior in case of missing fields.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
protected void setFailOnMissingField(boolean failOnMissingField) {
- this.failOnMissingField = failOnMissingField;
- }
-
- /**
- * Sets the mapping from table schema fields to JSON schema fields.
- *
- * @param fieldMapping The mapping from table schema fields to JSON schema fields.
- */
- protected void setFieldMapping(Map fieldMapping) {
- this.fieldMapping = fieldMapping;
+ ((JsonRowDeserializationSchema) getDeserializationSchema()).setFailOnMissingField(failOnMissingField);
}
//////// HELPER METHODS
- /** Converts the JSON schema into into the return type. */
- private static RowTypeInfo jsonSchemaToReturnType(TableSchema jsonSchema) {
- return new RowTypeInfo(jsonSchema.getTypes(), jsonSchema.getColumnNames());
- }
-
/**
* Abstract builder for a {@link KafkaJsonTableSource} to be extended by builders of subclasses of
* KafkaJsonTableSource.
*
* @param Type of the KafkaJsonTableSource produced by the builder.
* @param Type of the KafkaJsonTableSource.Builder subclass.
+ * @deprecated Use the {@link org.apache.flink.table.descriptors.Kafka} descriptor together
+ * with descriptors for schema and format instead. Descriptors allow for
+ * implementation-agnostic definition of tables. See also
+ * {@link org.apache.flink.table.api.TableEnvironment#from(ConnectorDescriptor)}.
*/
+ @Deprecated
protected abstract static class Builder
extends KafkaTableSource.Builder {
@@ -159,7 +116,9 @@ protected abstract static class Builder tableToJsonMapping) {
this.fieldMapping = tableToJsonMapping;
return builder();
@@ -188,7 +149,9 @@ public B withTableToJsonMapping(Map tableToJsonMapping) {
* field.
* If set to false, a missing field is set to null.
* @return The builder.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
public B failOnMissingField(boolean failOnMissingField) {
this.failOnMissingField = failOnMissingField;
return builder();
@@ -199,7 +162,9 @@ public B failOnMissingField(boolean failOnMissingField) {
* is returned.
*
* @return The JSON schema for the TableSource.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
protected TableSchema getJsonSchema() {
if (jsonSchema != null) {
return this.jsonSchema;
@@ -208,7 +173,13 @@ protected TableSchema getJsonSchema() {
}
}
- @Override
+ /**
+ * Configures a TableSource with optional parameters.
+ *
+ * @param source The TableSource to configure.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
+ */
+ @Deprecated
protected void configureTableSource(T source) {
super.configureTableSource(source);
// configure field mapping
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSourceFactory.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSourceFactory.java
deleted file mode 100644
index 98985694ab9a2..0000000000000
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSourceFactory.java
+++ /dev/null
@@ -1,96 +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.streaming.connectors.kafka;
-
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.formats.json.JsonSchemaConverter;
-import org.apache.flink.table.api.TableSchema;
-import org.apache.flink.table.descriptors.DescriptorProperties;
-import org.apache.flink.table.descriptors.FormatDescriptorValidator;
-import org.apache.flink.table.descriptors.JsonValidator;
-import org.apache.flink.table.descriptors.SchemaValidator;
-
-import java.util.Arrays;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-
-/**
- * Factory for creating configured instances of {@link KafkaJsonTableSource}.
- */
-public abstract class KafkaJsonTableSourceFactory extends KafkaTableSourceFactory {
-
- @Override
- protected String formatType() {
- return JsonValidator.FORMAT_TYPE_VALUE;
- }
-
- @Override
- protected int formatPropertyVersion() {
- return 1;
- }
-
- @Override
- protected List formatProperties() {
- return Arrays.asList(
- JsonValidator.FORMAT_JSON_SCHEMA,
- JsonValidator.FORMAT_SCHEMA,
- JsonValidator.FORMAT_FAIL_ON_MISSING_FIELD,
- FormatDescriptorValidator.FORMAT_DERIVE_SCHEMA());
- }
-
- @Override
- protected FormatDescriptorValidator formatValidator() {
- return new JsonValidator();
- }
-
- @Override
- protected KafkaTableSource.Builder createBuilderWithFormat(DescriptorProperties params) {
- final KafkaJsonTableSource.Builder builder = createKafkaJsonBuilder();
-
- // missing field
- params.getOptionalBoolean(JsonValidator.FORMAT_FAIL_ON_MISSING_FIELD)
- .ifPresent(builder::failOnMissingField);
-
- // json schema
- final TableSchema formatSchema;
- if (params.containsKey(JsonValidator.FORMAT_SCHEMA)) {
- final TypeInformation info = params.getType(JsonValidator.FORMAT_SCHEMA);
- formatSchema = TableSchema.fromTypeInfo(info);
- } else if (params.containsKey(JsonValidator.FORMAT_JSON_SCHEMA)) {
- final TypeInformation info = JsonSchemaConverter.convert(params.getString(JsonValidator.FORMAT_JSON_SCHEMA));
- formatSchema = TableSchema.fromTypeInfo(info);
- } else {
- formatSchema = SchemaValidator.deriveFormatFields(params);
- }
- builder.forJsonSchema(formatSchema);
-
- // field mapping
- final Map mapping = SchemaValidator.deriveFieldMapping(params, Optional.of(formatSchema));
- builder.withTableToJsonMapping(mapping);
-
- return builder;
- }
-
- /**
- * Creates a version specific {@link KafkaJsonTableSource.Builder}.
- */
- protected abstract KafkaJsonTableSource.Builder createKafkaJsonBuilder();
-
-}
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSource.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSource.java
index 29654b061c790..24cfd9436dbdd 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSource.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSource.java
@@ -28,6 +28,8 @@
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.api.Types;
import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.descriptors.ConnectorDescriptor;
+import org.apache.flink.table.sources.DefinedFieldMapping;
import org.apache.flink.table.sources.DefinedProctimeAttribute;
import org.apache.flink.table.sources.DefinedRowtimeAttributes;
import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
@@ -54,29 +56,40 @@
* override {@link #createKafkaConsumer(String, Properties, DeserializationSchema)}}.
*/
@Internal
-public abstract class KafkaTableSource
- implements StreamTableSource, DefinedProctimeAttribute, DefinedRowtimeAttributes {
+public abstract class KafkaTableSource implements
+ StreamTableSource,
+ DefinedProctimeAttribute,
+ DefinedRowtimeAttributes,
+ DefinedFieldMapping {
+
+ // common table source attributes
+ // TODO make all attributes final once we drop support for format-specific table sources
/** The schema of the table. */
private final TableSchema schema;
+ /** Field name of the processing time attribute, null if no processing time field is defined. */
+ private String proctimeAttribute;
+
+ /** Descriptor for a rowtime attribute. */
+ private List rowtimeAttributeDescriptors;
+
+ /** Mapping for the fields of the table schema to fields of the physical returned type or null. */
+ private Map fieldMapping;
+
+ // Kafka-specific attributes
+
/** The Kafka topic to consume. */
private final String topic;
/** Properties for the Kafka consumer. */
private final Properties properties;
- /** Type information describing the result type. */
- private TypeInformation returnType;
-
- /** Field name of the processing time attribute, null if no processing time field is defined. */
- private String proctimeAttribute;
-
- /** Descriptor for a rowtime attribute. */
- private List rowtimeAttributeDescriptors;
+ /** Deserialization schema for decoding records from Kafka. */
+ private final DeserializationSchema deserializationSchema;
/** The startup mode for the contained consumer (default is {@link StartupMode#GROUP_OFFSETS}). */
- private StartupMode startupMode = StartupMode.GROUP_OFFSETS;
+ private StartupMode startupMode;
/** Specific startup offsets; only relevant when startup mode is {@link StartupMode#SPECIFIC_OFFSETS}. */
private Map specificStartupOffsets;
@@ -84,21 +97,64 @@ public abstract class KafkaTableSource
/**
* Creates a generic Kafka {@link StreamTableSource}.
*
- * @param topic Kafka topic to consume.
- * @param properties Properties for the Kafka consumer.
- * @param schema Schema of the produced table.
- * @param returnType Type information of the produced physical DataStream.
+ * @param schema Schema of the produced table.
+ * @param proctimeAttribute Field name of the processing time attribute, null if no
+ * processing time field is defined.
+ * @param rowtimeAttributeDescriptors Descriptor for a rowtime attribute
+ * @param fieldMapping Mapping for the fields of the table schema to
+ * fields of the physical returned type or null.
+ * @param topic Kafka topic to consume.
+ * @param properties Properties for the Kafka consumer.
+ * @param deserializationSchema Deserialization schema for decoding records from Kafka.
+ * @param startupMode Startup mode for the contained consumer.
+ * @param specificStartupOffsets Specific startup offsets; only relevant when startup
+ * mode is {@link StartupMode#SPECIFIC_OFFSETS}.
*/
protected KafkaTableSource(
+ TableSchema schema,
+ String proctimeAttribute,
+ List rowtimeAttributeDescriptors,
+ Map fieldMapping,
String topic,
Properties properties,
- TableSchema schema,
- TypeInformation returnType) {
-
+ DeserializationSchema deserializationSchema,
+ StartupMode startupMode,
+ Map specificStartupOffsets) {
+ this.schema = Preconditions.checkNotNull(schema, "Schema must not be null.");
+ this.proctimeAttribute = validateProctimeAttribute(proctimeAttribute);
+ this.rowtimeAttributeDescriptors = validateRowtimeAttributeDescriptors(rowtimeAttributeDescriptors);
+ this.fieldMapping = fieldMapping;
this.topic = Preconditions.checkNotNull(topic, "Topic must not be null.");
this.properties = Preconditions.checkNotNull(properties, "Properties must not be null.");
- this.schema = Preconditions.checkNotNull(schema, "Schema must not be null.");
- this.returnType = Preconditions.checkNotNull(returnType, "Type information must not be null.");
+ this.deserializationSchema = Preconditions.checkNotNull(
+ deserializationSchema, "Deserialization schema must not be null.");
+ this.startupMode = Preconditions.checkNotNull(startupMode, "Startup mode must not be null.");
+ this.specificStartupOffsets = Preconditions.checkNotNull(
+ specificStartupOffsets, "Specific offsets must not be null.");
+ }
+
+ /**
+ * Creates a generic Kafka {@link StreamTableSource}.
+ *
+ * @param schema Schema of the produced table.
+ * @param topic Kafka topic to consume.
+ * @param properties Properties for the Kafka consumer.
+ * @param deserializationSchema Deserialization schema for decoding records from Kafka.
+ */
+ protected KafkaTableSource(
+ TableSchema schema,
+ String topic,
+ Properties properties,
+ DeserializationSchema deserializationSchema) {
+ this(
+ schema,
+ null,
+ Collections.emptyList(),
+ null,
+ topic, properties,
+ deserializationSchema,
+ StartupMode.GROUP_OFFSETS,
+ Collections.emptyMap());
}
/**
@@ -116,7 +172,7 @@ public DataStream getDataStream(StreamExecutionEnvironment env) {
@Override
public TypeInformation getReturnType() {
- return returnType;
+ return deserializationSchema.getProducedType();
}
@Override
@@ -134,34 +190,60 @@ public List getRowtimeAttributeDescriptors() {
return rowtimeAttributeDescriptors;
}
+ @Override
+ public Map getFieldMapping() {
+ return fieldMapping;
+ }
+
@Override
public String explainSource() {
return TableConnectorUtil.generateRuntimeName(this.getClass(), schema.getColumnNames());
}
+ /**
+ * Returns the properties for the Kafka consumer.
+ *
+ * @return properties for the Kafka consumer.
+ */
+ public Properties getProperties() {
+ return properties;
+ }
+
+ /**
+ * Returns the deserialization schema.
+ *
+ * @return The deserialization schema
+ */
+ public DeserializationSchema getDeserializationSchema(){
+ return deserializationSchema;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
- if (!(o instanceof KafkaTableSource)) {
+ // TODO force classes to be equal once we drop support for format-specific table sources
+ // if (o == null || getClass() != o.getClass()) {
+ if (o == null || !(o instanceof KafkaTableSource)) {
return false;
}
- KafkaTableSource that = (KafkaTableSource) o;
+ final KafkaTableSource that = (KafkaTableSource) o;
return Objects.equals(schema, that.schema) &&
- Objects.equals(topic, that.topic) &&
- Objects.equals(properties, that.properties) &&
- Objects.equals(returnType, that.returnType) &&
Objects.equals(proctimeAttribute, that.proctimeAttribute) &&
Objects.equals(rowtimeAttributeDescriptors, that.rowtimeAttributeDescriptors) &&
+ Objects.equals(fieldMapping, that.fieldMapping) &&
+ Objects.equals(topic, that.topic) &&
+ Objects.equals(properties, that.properties) &&
+ Objects.equals(deserializationSchema, that.deserializationSchema) &&
startupMode == that.startupMode &&
Objects.equals(specificStartupOffsets, that.specificStartupOffsets);
}
@Override
public int hashCode() {
- return Objects.hash(schema, topic, properties, returnType,
- proctimeAttribute, rowtimeAttributeDescriptors, startupMode, specificStartupOffsets);
+ return Objects.hash(schema, proctimeAttribute, rowtimeAttributeDescriptors, fieldMapping,
+ topic, properties, deserializationSchema, startupMode, specificStartupOffsets);
}
/**
@@ -195,14 +277,14 @@ protected FlinkKafkaConsumerBase getKafkaConsumer(
return kafkaConsumer;
}
- //////// SETTERS FOR OPTIONAL PARAMETERS
+ //////// VALIDATION FOR PARAMETERS
/**
- * Declares a field of the schema to be the processing time attribute.
+ * Validates a field of the schema to be the processing time attribute.
*
* @param proctimeAttribute The name of the field that becomes the processing time field.
*/
- protected void setProctimeAttribute(String proctimeAttribute) {
+ private String validateProctimeAttribute(String proctimeAttribute) {
if (proctimeAttribute != null) {
// validate that field exists and is of correct type
Option> tpe = schema.getType(proctimeAttribute);
@@ -212,15 +294,16 @@ protected void setProctimeAttribute(String proctimeAttribute) {
throw new ValidationException("Processing time attribute '" + proctimeAttribute + "' is not of type SQL_TIMESTAMP.");
}
}
- this.proctimeAttribute = proctimeAttribute;
+ return proctimeAttribute;
}
/**
- * Declares a list of fields to be rowtime attributes.
+ * Validates a list of fields to be rowtime attributes.
*
* @param rowtimeAttributeDescriptors The descriptors of the rowtime attributes.
*/
- protected void setRowtimeAttributeDescriptors(List rowtimeAttributeDescriptors) {
+ private List validateRowtimeAttributeDescriptors(List rowtimeAttributeDescriptors) {
+ Preconditions.checkNotNull(rowtimeAttributeDescriptors, "List of rowtime attributes must not be null.");
// validate that all declared fields exist and are of correct type
for (RowtimeAttributeDescriptor desc : rowtimeAttributeDescriptors) {
String rowtimeAttribute = desc.getAttributeName();
@@ -231,25 +314,64 @@ protected void setRowtimeAttributeDescriptors(List r
throw new ValidationException("Rowtime attribute '" + rowtimeAttribute + "' is not of type SQL_TIMESTAMP.");
}
}
- this.rowtimeAttributeDescriptors = rowtimeAttributeDescriptors;
+ return rowtimeAttributeDescriptors;
+ }
+
+ //////// SETTERS FOR OPTIONAL PARAMETERS
+
+ /**
+ * Declares a field of the schema to be the processing time attribute.
+ *
+ * @param proctimeAttribute The name of the field that becomes the processing time field.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
+ */
+ @Deprecated
+ protected void setProctimeAttribute(String proctimeAttribute) {
+ this.proctimeAttribute = validateProctimeAttribute(proctimeAttribute);
+ }
+
+ /**
+ * Declares a list of fields to be rowtime attributes.
+ *
+ * @param rowtimeAttributeDescriptors The descriptors of the rowtime attributes.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
+ */
+ @Deprecated
+ protected void setRowtimeAttributeDescriptors(List rowtimeAttributeDescriptors) {
+ this.rowtimeAttributeDescriptors = validateRowtimeAttributeDescriptors(rowtimeAttributeDescriptors);
}
/**
* Sets the startup mode of the TableSource.
*
* @param startupMode The startup mode.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
protected void setStartupMode(StartupMode startupMode) {
- this.startupMode = startupMode;
+ this.startupMode = Preconditions.checkNotNull(startupMode);
}
/**
* Sets the startup offsets of the TableSource; only relevant when the startup mode is {@link StartupMode#SPECIFIC_OFFSETS}.
*
* @param specificStartupOffsets The startup offsets for different partitions.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
protected void setSpecificStartupOffsets(Map specificStartupOffsets) {
- this.specificStartupOffsets = specificStartupOffsets;
+ this.specificStartupOffsets = Preconditions.checkNotNull(specificStartupOffsets);
+ }
+
+ /**
+ * Mapping for the fields of the table schema to fields of the physical returned type or null.
+ *
+ * @param fieldMapping The mapping from table schema fields to format schema fields.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
+ */
+ @Deprecated
+ protected void setFieldMapping(Map fieldMapping) {
+ this.fieldMapping = fieldMapping;
}
//////// ABSTRACT METHODS FOR SUBCLASSES
@@ -267,20 +389,18 @@ protected abstract FlinkKafkaConsumerBase createKafkaConsumer(
Properties properties,
DeserializationSchema deserializationSchema);
- /**
- * Returns the deserialization schema.
- *
- * @return The deserialization schema
- */
- protected abstract DeserializationSchema getDeserializationSchema();
-
/**
* Abstract builder for a {@link KafkaTableSource} to be extended by builders of subclasses of
* KafkaTableSource.
*
* @param Type of the KafkaTableSource produced by the builder.
* @param Type of the KafkaTableSource.Builder subclass.
+ * @deprecated Use the {@link org.apache.flink.table.descriptors.Kafka} descriptor together
+ * with descriptors for schema and format instead. Descriptors allow for
+ * implementation-agnostic definition of tables. See also
+ * {@link org.apache.flink.table.api.TableEnvironment#from(ConnectorDescriptor)}.
*/
+ @Deprecated
protected abstract static class Builder {
private String topic;
@@ -299,13 +419,14 @@ protected abstract static class Builder specificStartupOffsets = null;
-
/**
* Sets the topic from which the table is read.
*
* @param topic The topic from which the table is read.
* @return The builder.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
public B forTopic(String topic) {
Preconditions.checkNotNull(topic, "Topic must not be null.");
Preconditions.checkArgument(this.topic == null, "Topic has already been set.");
@@ -318,7 +439,9 @@ public B forTopic(String topic) {
*
* @param props The configuration properties for the Kafka consumer.
* @return The builder.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
public B withKafkaProperties(Properties props) {
Preconditions.checkNotNull(props, "Properties must not be null.");
Preconditions.checkArgument(this.kafkaProps == null, "Properties have already been set.");
@@ -331,7 +454,9 @@ public B withKafkaProperties(Properties props) {
*
* @param schema The schema of the produced table.
* @return The builder.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
public B withSchema(TableSchema schema) {
Preconditions.checkNotNull(schema, "Schema must not be null.");
Preconditions.checkArgument(this.schema == null, "Schema has already been set.");
@@ -345,7 +470,9 @@ public B withSchema(TableSchema schema) {
*
* @param proctimeAttribute The name of the processing time attribute in the table schema.
* @return The builder.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
public B withProctimeAttribute(String proctimeAttribute) {
Preconditions.checkNotNull(proctimeAttribute, "Proctime attribute must not be null.");
Preconditions.checkArgument(!proctimeAttribute.isEmpty(), "Proctime attribute must not be empty.");
@@ -362,7 +489,9 @@ public B withProctimeAttribute(String proctimeAttribute) {
* @param timestampExtractor The {@link TimestampExtractor} to extract the rowtime attribute from the physical type.
* @param watermarkStrategy The {@link WatermarkStrategy} to generate watermarks for the rowtime attribute.
* @return The builder.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
public B withRowtimeAttribute(
String rowtimeAttribute,
TimestampExtractor timestampExtractor,
@@ -389,7 +518,9 @@ public B withRowtimeAttribute(
* @param rowtimeAttribute The name of the rowtime attribute in the table schema.
* @param watermarkStrategy The {@link WatermarkStrategy} to generate watermarks for the rowtime attribute.
* @return The builder.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
public B withKafkaTimestampAsRowtimeAttribute(
String rowtimeAttribute,
WatermarkStrategy watermarkStrategy) {
@@ -412,7 +543,9 @@ public B withKafkaTimestampAsRowtimeAttribute(
* Configures the TableSource to start reading from the earliest offset for all partitions.
*
* @see FlinkKafkaConsumerBase#setStartFromEarliest()
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
public B fromEarliest() {
this.startupMode = StartupMode.EARLIEST;
this.specificStartupOffsets = null;
@@ -423,7 +556,9 @@ public B fromEarliest() {
* Configures the TableSource to start reading from the latest offset for all partitions.
*
* @see FlinkKafkaConsumerBase#setStartFromLatest()
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
public B fromLatest() {
this.startupMode = StartupMode.LATEST;
this.specificStartupOffsets = null;
@@ -434,7 +569,9 @@ public B fromLatest() {
* Configures the TableSource to start reading from any committed group offsets found in Zookeeper / Kafka brokers.
*
* @see FlinkKafkaConsumerBase#setStartFromGroupOffsets()
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
public B fromGroupOffsets() {
this.startupMode = StartupMode.GROUP_OFFSETS;
this.specificStartupOffsets = null;
@@ -446,7 +583,9 @@ public B fromGroupOffsets() {
*
* @param specificStartupOffsets the specified offsets for partitions
* @see FlinkKafkaConsumerBase#setStartFromSpecificOffsets(Map)
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
public B fromSpecificOffsets(Map specificStartupOffsets) {
this.startupMode = StartupMode.SPECIFIC_OFFSETS;
this.specificStartupOffsets = Preconditions.checkNotNull(specificStartupOffsets);
@@ -457,7 +596,9 @@ public B fromSpecificOffsets(Map specificStartupOffse
* Returns the configured topic.
*
* @return the configured topic.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
protected String getTopic() {
return this.topic;
}
@@ -466,7 +607,9 @@ protected String getTopic() {
* Returns the configured Kafka properties.
*
* @return the configured Kafka properties.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
protected Properties getKafkaProps() {
return this.kafkaProps;
}
@@ -475,7 +618,9 @@ protected Properties getKafkaProps() {
* Returns the configured table schema.
*
* @return the configured table schema.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
protected TableSchema getTableSchema() {
return this.schema;
}
@@ -484,14 +629,18 @@ protected TableSchema getTableSchema() {
* True if the KafkaSource supports Kafka timestamps, false otherwise.
*
* @return True if the KafkaSource supports Kafka timestamps, false otherwise.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
protected abstract boolean supportsKafkaTimestamps();
/**
* Configures a TableSource with optional parameters.
*
* @param tableSource The TableSource to configure.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
protected void configureTableSource(T tableSource) {
// configure processing time attributes
tableSource.setProctimeAttribute(proctimeAttribute);
@@ -516,13 +665,20 @@ protected void configureTableSource(T tableSource) {
/**
* Returns the builder.
* @return the builder.
+ * @deprecated Use the {@link org.apache.flink.table.descriptors.Kafka} descriptor together
+ * with descriptors for schema and format instead. Descriptors allow for
+ * implementation-agnostic definition of tables. See also
+ * {@link org.apache.flink.table.api.TableEnvironment#from(ConnectorDescriptor)}.
*/
+ @Deprecated
protected abstract B builder();
/**
* Builds the configured {@link KafkaTableSource}.
* @return The configured {@link KafkaTableSource}.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
protected abstract KafkaTableSource build();
}
}
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceFactory.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceFactory.java
index 6aaef9b8d2b45..3a2a81e5fe62c 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceFactory.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceFactory.java
@@ -18,13 +18,16 @@
package org.apache.flink.streaming.connectors.kafka;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
import org.apache.flink.table.api.TableException;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.descriptors.DescriptorProperties;
-import org.apache.flink.table.descriptors.FormatDescriptorValidator;
import org.apache.flink.table.descriptors.KafkaValidator;
import org.apache.flink.table.descriptors.SchemaValidator;
+import org.apache.flink.table.formats.DeserializationSchemaFactory;
+import org.apache.flink.table.formats.TableFormatFactoryService;
import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
import org.apache.flink.table.sources.TableSource;
import org.apache.flink.table.sources.TableSourceFactory;
@@ -35,13 +38,13 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Properties;
import static org.apache.flink.table.descriptors.ConnectorDescriptorValidator.CONNECTOR_PROPERTY_VERSION;
import static org.apache.flink.table.descriptors.ConnectorDescriptorValidator.CONNECTOR_TYPE;
import static org.apache.flink.table.descriptors.ConnectorDescriptorValidator.CONNECTOR_VERSION;
-import static org.apache.flink.table.descriptors.FormatDescriptorValidator.FORMAT_PROPERTY_VERSION;
-import static org.apache.flink.table.descriptors.FormatDescriptorValidator.FORMAT_TYPE;
+import static org.apache.flink.table.descriptors.FormatDescriptorValidator.FORMAT;
import static org.apache.flink.table.descriptors.KafkaValidator.CONNECTOR_PROPERTIES;
import static org.apache.flink.table.descriptors.KafkaValidator.CONNECTOR_PROPERTIES_KEY;
import static org.apache.flink.table.descriptors.KafkaValidator.CONNECTOR_PROPERTIES_VALUE;
@@ -66,21 +69,16 @@
import static org.apache.flink.table.descriptors.SchemaValidator.SCHEMA_TYPE;
/**
- * Factory for creating configured instances of {@link KafkaJsonTableSource}.
+ * Factory for creating configured instances of {@link KafkaTableSource}.
*/
-abstract class KafkaTableSourceFactory implements TableSourceFactory {
+public abstract class KafkaTableSourceFactory implements TableSourceFactory {
@Override
public Map requiredContext() {
Map context = new HashMap<>();
context.put(CONNECTOR_TYPE(), CONNECTOR_TYPE_VALUE_KAFKA); // kafka
context.put(CONNECTOR_VERSION(), kafkaVersion()); // version
-
- context.put(FORMAT_TYPE(), formatType()); // format
-
context.put(CONNECTOR_PROPERTY_VERSION(), "1"); // backwards compatibility
- context.put(FORMAT_PROPERTY_VERSION(), String.valueOf(formatPropertyVersion()));
-
return context;
}
@@ -113,7 +111,8 @@ public List supportedProperties() {
properties.add(SCHEMA() + ".#." + ROWTIME_WATERMARKS_SERIALIZED());
properties.add(SCHEMA() + ".#." + ROWTIME_WATERMARKS_DELAY());
- properties.addAll(formatProperties());
+ // format wildcard
+ properties.add(FORMAT() + ".*");
return properties;
}
@@ -124,110 +123,127 @@ public TableSource create(Map properties) {
params.putProperties(properties);
// validate
- new SchemaValidator(true).validate(params);
+ // allow Kafka timestamps to be used, watermarks can not be received from source
+ new SchemaValidator(true, supportsKafkaTimestamps(), false).validate(params);
new KafkaValidator().validate(params);
- formatValidator().validate(params);
- // build
- final KafkaTableSource.Builder builder = createBuilderWithFormat(params);
+ // deserialization schema using format discovery
+ final DeserializationSchemaFactory formatFactory = TableFormatFactoryService.find(
+ DeserializationSchemaFactory.class,
+ properties,
+ this.getClass().getClassLoader());
+ @SuppressWarnings("unchecked")
+ final DeserializationSchema deserializationSchema = (DeserializationSchema) formatFactory
+ .createDeserializationSchema(properties);
- // topic
- final String topic = params.getString(CONNECTOR_TOPIC);
- builder.forTopic(topic);
+ // schema
+ final TableSchema schema = params.getTableSchema(SCHEMA());
+
+ // proctime
+ final String proctimeAttribute = SchemaValidator.deriveProctimeAttribute(params).orElse(null);
+
+ // rowtime
+ final List rowtimeAttributes = SchemaValidator.deriveRowtimeAttributes(params);
+
+ // field mapping
+ final Map fieldMapping = SchemaValidator.deriveFieldMapping(params, Optional.of(schema));
// properties
- final Properties props = new Properties();
+ final Properties kafkaProperties = new Properties();
final List> propsList = params.getFixedIndexedProperties(
CONNECTOR_PROPERTIES,
Arrays.asList(CONNECTOR_PROPERTIES_KEY, CONNECTOR_PROPERTIES_VALUE));
- propsList.forEach(kv -> props.put(
+ propsList.forEach(kv -> kafkaProperties.put(
params.getString(kv.get(CONNECTOR_PROPERTIES_KEY)),
params.getString(kv.get(CONNECTOR_PROPERTIES_VALUE))
));
- builder.withKafkaProperties(props);
+
+ // topic
+ final String topic = params.getString(CONNECTOR_TOPIC);
// startup mode
- params
+ final Map specificOffsets = new HashMap<>();
+ final StartupMode startupMode = params
.getOptionalString(CONNECTOR_STARTUP_MODE)
- .ifPresent(startupMode -> {
- switch (startupMode) {
-
- case KafkaValidator.CONNECTOR_STARTUP_MODE_VALUE_EARLIEST:
- builder.fromEarliest();
- break;
-
- case KafkaValidator.CONNECTOR_STARTUP_MODE_VALUE_LATEST:
- builder.fromLatest();
- break;
-
- case KafkaValidator.CONNECTOR_STARTUP_MODE_VALUE_GROUP_OFFSETS:
- builder.fromGroupOffsets();
- break;
-
- case KafkaValidator.CONNECTOR_STARTUP_MODE_VALUE_SPECIFIC_OFFSETS:
- final Map offsetMap = new HashMap<>();
-
- final List> offsetList = params.getFixedIndexedProperties(
- CONNECTOR_SPECIFIC_OFFSETS,
- Arrays.asList(CONNECTOR_SPECIFIC_OFFSETS_PARTITION, CONNECTOR_SPECIFIC_OFFSETS_OFFSET));
- offsetList.forEach(kv -> {
- final int partition = params.getInt(kv.get(CONNECTOR_SPECIFIC_OFFSETS_PARTITION));
- final long offset = params.getLong(kv.get(CONNECTOR_SPECIFIC_OFFSETS_OFFSET));
- final KafkaTopicPartition topicPartition = new KafkaTopicPartition(topic, partition);
- offsetMap.put(topicPartition, offset);
- });
- builder.fromSpecificOffsets(offsetMap);
- break;
+ .map(modeString -> {
+ switch (modeString) {
+ case KafkaValidator.CONNECTOR_STARTUP_MODE_VALUE_EARLIEST:
+ return StartupMode.EARLIEST;
+
+ case KafkaValidator.CONNECTOR_STARTUP_MODE_VALUE_LATEST:
+ return StartupMode.LATEST;
+
+ case KafkaValidator.CONNECTOR_STARTUP_MODE_VALUE_GROUP_OFFSETS:
+ return StartupMode.GROUP_OFFSETS;
+
+ case KafkaValidator.CONNECTOR_STARTUP_MODE_VALUE_SPECIFIC_OFFSETS:
+ final List> offsetList = params.getFixedIndexedProperties(
+ CONNECTOR_SPECIFIC_OFFSETS,
+ Arrays.asList(CONNECTOR_SPECIFIC_OFFSETS_PARTITION, CONNECTOR_SPECIFIC_OFFSETS_OFFSET));
+ offsetList.forEach(kv -> {
+ final int partition = params.getInt(kv.get(CONNECTOR_SPECIFIC_OFFSETS_PARTITION));
+ final long offset = params.getLong(kv.get(CONNECTOR_SPECIFIC_OFFSETS_OFFSET));
+ final KafkaTopicPartition topicPartition = new KafkaTopicPartition(topic, partition);
+ specificOffsets.put(topicPartition, offset);
+ });
+ return StartupMode.SPECIFIC_OFFSETS;
+ default:
+ throw new TableException("Unsupported startup mode. Validator should have checked that.");
}
- });
-
- // schema
- final TableSchema schema = params.getTableSchema(SCHEMA());
- builder.withSchema(schema);
-
- // proctime
- SchemaValidator.deriveProctimeAttribute(params).ifPresent(builder::withProctimeAttribute);
-
- // rowtime
- final List descriptors = SchemaValidator.deriveRowtimeAttributes(params);
- if (descriptors.size() > 1) {
- throw new TableException("More than one rowtime attribute is not supported yet.");
- } else if (descriptors.size() == 1) {
- final RowtimeAttributeDescriptor desc = descriptors.get(0);
- builder.withRowtimeAttribute(desc.getAttributeName(), desc.getTimestampExtractor(), desc.getWatermarkStrategy());
- }
-
- return builder.build();
+ }).orElse(StartupMode.GROUP_OFFSETS);
+
+ return createKafkaTableSource(
+ schema,
+ proctimeAttribute,
+ rowtimeAttributes,
+ fieldMapping,
+ topic,
+ kafkaProperties,
+ deserializationSchema,
+ startupMode,
+ specificOffsets);
}
- /**
- * Returns the format type string (e.g., "json").
- */
- protected abstract String formatType();
+ // --------------------------------------------------------------------------------------------
+ // For version-specific factories
+ // --------------------------------------------------------------------------------------------
/**
- * Returns the format property version.
- */
- protected abstract int formatPropertyVersion();
-
- /**
- * Returns the properties of the format.
- */
- protected abstract List formatProperties();
-
- /**
- * Returns the validator for the format.
+ * Returns the Kafka version.
*/
- protected abstract FormatDescriptorValidator formatValidator();
+ protected abstract String kafkaVersion();
/**
- * Returns the Kafka version.
+ * True if the Kafka source supports Kafka timestamps, false otherwise.
+ *
+ * @return True if the Kafka source supports Kafka timestamps, false otherwise.
*/
- protected abstract String kafkaVersion();
+ protected abstract boolean supportsKafkaTimestamps();
/**
- * Creates a builder with all the format-related configurations have been set.
+ * Constructs the version-specific Kafka table source.
+ *
+ * @param schema Schema of the produced table.
+ * @param proctimeAttribute Field name of the processing time attribute, null if no
+ * processing time field is defined.
+ * @param rowtimeAttributeDescriptors Descriptor for a rowtime attribute
+ * @param fieldMapping Mapping for the fields of the table schema to
+ * fields of the physical returned type or null.
+ * @param topic Kafka topic to consume.
+ * @param properties Properties for the Kafka consumer.
+ * @param deserializationSchema Deserialization schema for decoding records from Kafka.
+ * @param startupMode Startup mode for the contained consumer.
+ * @param specificStartupOffsets Specific startup offsets; only relevant when startup
+ * mode is {@link StartupMode#SPECIFIC_OFFSETS}.
*/
- protected abstract KafkaTableSource.Builder createBuilderWithFormat(DescriptorProperties params);
+ protected abstract KafkaTableSource createKafkaTableSource(
+ TableSchema schema,
+ String proctimeAttribute,
+ List rowtimeAttributeDescriptors,
+ Map fieldMapping,
+ String topic, Properties properties,
+ DeserializationSchema deserializationSchema,
+ StartupMode startupMode,
+ Map specificStartupOffsets);
}
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaAvroTableSourceFactoryTestBase.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaAvroTableSourceFactoryTestBase.java
deleted file mode 100644
index 95525590777b3..0000000000000
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaAvroTableSourceFactoryTestBase.java
+++ /dev/null
@@ -1,123 +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.streaming.connectors.kafka;
-
-import org.apache.flink.api.common.typeinfo.Types;
-import org.apache.flink.formats.avro.generated.Address;
-import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
-import org.apache.flink.table.api.TableSchema;
-import org.apache.flink.table.descriptors.Avro;
-import org.apache.flink.table.descriptors.FormatDescriptor;
-import org.apache.flink.table.descriptors.Kafka;
-import org.apache.flink.table.descriptors.Schema;
-import org.apache.flink.table.descriptors.TestTableSourceDescriptor;
-import org.apache.flink.table.sources.TableSource;
-import org.apache.flink.table.sources.TableSourceFactoryService;
-
-import org.junit.Test;
-
-import static org.junit.Assert.assertEquals;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Properties;
-
-/**
- * Tests for {@link KafkaAvroTableSourceFactory}.
- */
-public abstract class KafkaAvroTableSourceFactoryTestBase {
-
- private static final String TOPIC = "test-topic";
-
- protected abstract String version();
-
- protected abstract KafkaAvroTableSource.Builder builder();
-
- @Test
- public void testTableSourceFromAvroSchema() {
- testTableSource(new Avro().recordClass(Address.class));
- }
-
- private void testTableSource(FormatDescriptor format) {
- // construct table source using a builder
-
- final Map tableAvroMapping = new HashMap<>();
- tableAvroMapping.put("a_street", "street");
- tableAvroMapping.put("street", "street");
- tableAvroMapping.put("b_city", "city");
- tableAvroMapping.put("city", "city");
- tableAvroMapping.put("c_state", "state");
- tableAvroMapping.put("state", "state");
- tableAvroMapping.put("zip", "zip");
- tableAvroMapping.put("num", "num");
-
- final Properties props = new Properties();
- props.put("group.id", "test-group");
- props.put("bootstrap.servers", "localhost:1234");
-
- final Map specificOffsets = new HashMap<>();
- specificOffsets.put(new KafkaTopicPartition(TOPIC, 0), 100L);
- specificOffsets.put(new KafkaTopicPartition(TOPIC, 1), 123L);
-
- KafkaAvroTableSource.Builder builder = builder();
-
- builder.forAvroRecordClass(Address.class)
- .withTableToAvroMapping(tableAvroMapping);
-
- final KafkaTableSource builderSource = builder
- .withKafkaProperties(props)
- .forTopic(TOPIC)
- .fromSpecificOffsets(specificOffsets)
- .withSchema(
- TableSchema.builder()
- .field("a_street", Types.STRING)
- .field("b_city", Types.STRING)
- .field("c_state", Types.STRING)
- .field("zip", Types.STRING)
- .field("proctime", Types.SQL_TIMESTAMP)
- .build())
- .withProctimeAttribute("proctime")
- .build();
-
- // construct table source using descriptors and table source factory
-
- final Map offsets = new HashMap<>();
- offsets.put(0, 100L);
- offsets.put(1, 123L);
-
- final TestTableSourceDescriptor testDesc = new TestTableSourceDescriptor(
- new Kafka()
- .version(version())
- .topic(TOPIC)
- .properties(props)
- .startFromSpecificOffsets(offsets))
- .addFormat(format)
- .addSchema(
- new Schema()
- .field("a_street", Types.STRING).from("street")
- .field("b_city", Types.STRING).from("city")
- .field("c_state", Types.STRING).from("state")
- .field("zip", Types.STRING)
- .field("proctime", Types.SQL_TIMESTAMP).proctime());
-
- final TableSource factorySource = TableSourceFactoryService.findAndCreateTableSource(testDesc);
-
- assertEquals(builderSource, factorySource);
- }
-}
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaAvroTableSourceTestBase.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaAvroTableSourceTestBase.java
index f86fc9542e143..140ce21d9421b 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaAvroTableSourceTestBase.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaAvroTableSourceTestBase.java
@@ -35,8 +35,12 @@
/**
* Abstract test base for all Kafka Avro table sources.
+ *
+ * @deprecated Ensures backwards compatibility with Flink 1.5. Can be removed once we
+ * drop support for format-specific table sources.
*/
-public abstract class KafkaAvroTableSourceTestBase extends KafkaTableSourceTestBase {
+@Deprecated
+public abstract class KafkaAvroTableSourceTestBase extends KafkaTableSourceBuilderTestBase {
@Override
protected void configureBuilder(KafkaTableSource.Builder builder) {
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSourceFactoryTestBase.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSourceFactoryTestBase.java
index e50810762c698..84c5bd30099d5 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSourceFactoryTestBase.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSourceFactoryTestBase.java
@@ -19,7 +19,7 @@
package org.apache.flink.streaming.connectors.kafka;
import org.apache.flink.api.common.typeinfo.Types;
-import org.apache.flink.formats.json.JsonSchemaConverter;
+import org.apache.flink.formats.json.JsonRowSchemaConverter;
import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.descriptors.FormatDescriptor;
@@ -31,7 +31,7 @@
import org.apache.flink.table.sources.TableSource;
import org.apache.flink.table.sources.TableSourceFactoryService;
import org.apache.flink.table.sources.tsextractors.ExistingField;
-import org.apache.flink.table.sources.wmstrategies.PreserveWatermarks;
+import org.apache.flink.table.sources.wmstrategies.AscendingTimestamps;
import org.junit.Test;
@@ -42,8 +42,12 @@
import static org.junit.Assert.assertEquals;
/**
- * Tests for {@link KafkaJsonTableSourceFactory}.
+ * Tests for legacy KafkaJsonTableSourceFactory.
+ *
+ * @deprecated Ensures backwards compatibility with Flink 1.5. Can be removed once we
+ * drop support for format-specific table sources.
*/
+@Deprecated
public abstract class KafkaJsonTableSourceFactoryTestBase {
private static final String JSON_SCHEMA =
@@ -94,10 +98,8 @@ private void testTableSource(FormatDescriptor format) {
// construct table source using a builder
final Map tableJsonMapping = new HashMap<>();
- tableJsonMapping.put("name", "name");
tableJsonMapping.put("fruit-name", "name");
tableJsonMapping.put("count", "count");
- tableJsonMapping.put("time", "time");
final Properties props = new Properties();
props.put("group.id", "test-group");
@@ -108,7 +110,7 @@ private void testTableSource(FormatDescriptor format) {
specificOffsets.put(new KafkaTopicPartition(TOPIC, 1), 123L);
final KafkaTableSource builderSource = builder()
- .forJsonSchema(TableSchema.fromTypeInfo(JsonSchemaConverter.convert(JSON_SCHEMA)))
+ .forJsonSchema(TableSchema.fromTypeInfo(JsonRowSchemaConverter.convert(JSON_SCHEMA)))
.failOnMissingField(true)
.withTableToJsonMapping(tableJsonMapping)
.withKafkaProperties(props)
@@ -122,7 +124,7 @@ private void testTableSource(FormatDescriptor format) {
.field("proc-time", Types.SQL_TIMESTAMP)
.build())
.withProctimeAttribute("proc-time")
- .withRowtimeAttribute("event-time", new ExistingField("time"), PreserveWatermarks.INSTANCE())
+ .withRowtimeAttribute("event-time", new ExistingField("time"), new AscendingTimestamps())
.build();
// construct table source using descriptors and table source factory
@@ -143,7 +145,7 @@ private void testTableSource(FormatDescriptor format) {
.field("fruit-name", Types.STRING).from("name")
.field("count", Types.BIG_DEC) // no from so it must match with the input
.field("event-time", Types.SQL_TIMESTAMP).rowtime(
- new Rowtime().timestampsFromField("time").watermarksFromSource())
+ new Rowtime().timestampsFromField("time").watermarksPeriodicAscending())
.field("proc-time", Types.SQL_TIMESTAMP).proctime());
final TableSource factorySource = TableSourceFactoryService.findAndCreateTableSource(testDesc);
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSourceTestBase.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSourceTestBase.java
index 9b867f6afdbae..4f95b96293690 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSourceTestBase.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaJsonTableSourceTestBase.java
@@ -35,8 +35,12 @@
/**
* Abstract test base for all Kafka JSON table sources.
+ *
+ * @deprecated Ensures backwards compatibility with Flink 1.5. Can be removed once we
+ * drop support for format-specific table sources.
*/
-public abstract class KafkaJsonTableSourceTestBase extends KafkaTableSourceTestBase {
+@Deprecated
+public abstract class KafkaJsonTableSourceTestBase extends KafkaTableSourceBuilderTestBase {
@Test
public void testJsonEqualsTableSchema() {
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceTestBase.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceBuilderTestBase.java
similarity index 96%
rename from flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceTestBase.java
rename to flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceBuilderTestBase.java
index 94281dfb430a9..b9ae47b3f5413 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceTestBase.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceBuilderTestBase.java
@@ -50,9 +50,13 @@
import static org.mockito.Mockito.when;
/**
- * Abstract test base for all Kafka table sources.
+ * Abstract test base for all format-specific Kafka table sources with builders.
+ *
+ * @deprecated Ensures backwards compatibility with Flink 1.5. Can be removed once we
+ * drop support for format-specific table sources.
*/
-public abstract class KafkaTableSourceTestBase {
+@Deprecated
+public abstract class KafkaTableSourceBuilderTestBase {
static final String[] FIELD_NAMES =
new String[]{"field1", "field2", "time1", "time2", "field3"};
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceFactoryTestBase.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceFactoryTestBase.java
new file mode 100644
index 0000000000000..9e1b58f9f78fa
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceFactoryTestBase.java
@@ -0,0 +1,187 @@
+/*
+ * 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.streaming.connectors.kafka;
+
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.Types;
+import org.apache.flink.table.descriptors.Kafka;
+import org.apache.flink.table.descriptors.Rowtime;
+import org.apache.flink.table.descriptors.Schema;
+import org.apache.flink.table.descriptors.TestTableSourceDescriptor;
+import org.apache.flink.table.formats.utils.TestDeserializationSchema;
+import org.apache.flink.table.formats.utils.TestTableFormat;
+import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.table.sources.TableSourceFactoryService;
+import org.apache.flink.table.sources.tsextractors.ExistingField;
+import org.apache.flink.table.sources.wmstrategies.AscendingTimestamps;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Abstract test base for {@link KafkaTableSourceFactory}.
+ */
+public abstract class KafkaTableSourceFactoryTestBase extends TestLogger {
+
+ private static final String TOPIC = "myTopic";
+ private static final int PARTITION_0 = 0;
+ private static final long OFFSET_0 = 100L;
+ private static final int PARTITION_1 = 1;
+ private static final long OFFSET_1 = 123L;
+ private static final String FRUIT_NAME = "fruit-name";
+ private static final String NAME = "name";
+ private static final String COUNT = "count";
+ private static final String TIME = "time";
+ private static final String EVENT_TIME = "event-time";
+ private static final String PROC_TIME = "proc-time";
+ private static final Properties KAFKA_PROPERTIES = new Properties();
+ static {
+ KAFKA_PROPERTIES.setProperty("zookeeper.connect", "dummy");
+ KAFKA_PROPERTIES.setProperty("group.id", "dummy");
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testTableSource() {
+
+ // prepare parameters for Kafka table source
+
+ final TableSchema schema = TableSchema.builder()
+ .field(FRUIT_NAME, Types.STRING())
+ .field(COUNT, Types.DECIMAL())
+ .field(EVENT_TIME, Types.SQL_TIMESTAMP())
+ .field(PROC_TIME, Types.SQL_TIMESTAMP())
+ .build();
+
+ final List rowtimeAttributeDescriptors = Collections.singletonList(
+ new RowtimeAttributeDescriptor(EVENT_TIME, new ExistingField(TIME), new AscendingTimestamps()));
+
+ final Map fieldMapping = new HashMap<>();
+ fieldMapping.put(FRUIT_NAME, NAME);
+ fieldMapping.put(COUNT, COUNT);
+
+ final Map specificOffsets = new HashMap<>();
+ specificOffsets.put(new KafkaTopicPartition(TOPIC, PARTITION_0), OFFSET_0);
+ specificOffsets.put(new KafkaTopicPartition(TOPIC, PARTITION_1), OFFSET_1);
+
+ final TestDeserializationSchema deserializationSchema = new TestDeserializationSchema(
+ TableSchema.builder()
+ .field(NAME, Types.STRING())
+ .field(COUNT, Types.DECIMAL())
+ .field(TIME, Types.SQL_TIMESTAMP())
+ .build()
+ .toRowType()
+ );
+
+ final StartupMode startupMode = StartupMode.SPECIFIC_OFFSETS;
+
+ final KafkaTableSource expected = getExpectedKafkaTableSource(
+ schema, PROC_TIME,
+ rowtimeAttributeDescriptors,
+ fieldMapping,
+ TOPIC,
+ KAFKA_PROPERTIES,
+ deserializationSchema,
+ startupMode,
+ specificOffsets);
+
+ // construct table source using descriptors and table source factory
+
+ final Map offsets = new HashMap<>();
+ offsets.put(PARTITION_0, OFFSET_0);
+ offsets.put(PARTITION_1, OFFSET_1);
+
+ final TestTableSourceDescriptor testDesc = new TestTableSourceDescriptor(
+ new Kafka()
+ .version(getKafkaVersion())
+ .topic(TOPIC)
+ .properties(KAFKA_PROPERTIES)
+ .startFromSpecificOffsets(offsets))
+ .addFormat(new TestTableFormat())
+ .addSchema(
+ new Schema()
+ .field(FRUIT_NAME, Types.STRING()).from(NAME)
+ .field(COUNT, Types.DECIMAL()) // no from so it must match with the input
+ .field(EVENT_TIME, Types.SQL_TIMESTAMP()).rowtime(
+ new Rowtime().timestampsFromField(TIME).watermarksPeriodicAscending())
+ .field(PROC_TIME, Types.SQL_TIMESTAMP()).proctime());
+
+ final TableSource actualSource = TableSourceFactoryService.findAndCreateTableSource(testDesc);
+
+ assertEquals(expected, actualSource);
+
+ // test Kafka consumer
+ final KafkaTableSource actualKafkaSource = (KafkaTableSource) actualSource;
+ final StreamExecutionEnvironmentMock mock = new StreamExecutionEnvironmentMock();
+ actualKafkaSource.getDataStream(mock);
+ assertTrue(getExpectedFlinkKafkaConsumer().isAssignableFrom(mock.function.getClass()));
+ }
+
+ private static class StreamExecutionEnvironmentMock extends StreamExecutionEnvironment {
+
+ public SourceFunction function;
+
+ @Override
+ public DataStreamSource addSource(SourceFunction function) {
+ this.function = function;
+ return super.addSource(function);
+ }
+
+ @Override
+ public JobExecutionResult execute(String jobName) {
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // For version-specific tests
+ // --------------------------------------------------------------------------------------------
+
+ protected abstract String getKafkaVersion();
+
+ protected abstract Class> getExpectedFlinkKafkaConsumer();
+
+ protected abstract KafkaTableSource getExpectedKafkaTableSource(
+ TableSchema schema,
+ String proctimeAttribute,
+ List rowtimeAttributeDescriptors,
+ Map fieldMapping,
+ String topic, Properties properties,
+ DeserializationSchema deserializationSchema,
+ StartupMode startupMode,
+ Map specificStartupOffsets);
+}
diff --git a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowDeserializationSchema.java b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowDeserializationSchema.java
index c36a4be13b54e..5e3edf77a5e8a 100644
--- a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowDeserializationSchema.java
+++ b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowDeserializationSchema.java
@@ -60,6 +60,7 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.TimeZone;
/**
@@ -175,6 +176,24 @@ public TypeInformation getProducedType() {
return typeInfo;
}
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ final AvroRowDeserializationSchema that = (AvroRowDeserializationSchema) o;
+ return Objects.equals(recordClazz, that.recordClazz) &&
+ Objects.equals(schemaString, that.schemaString);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(recordClazz, schemaString);
+ }
+
// --------------------------------------------------------------------------------------------
private Row convertAvroRecordToRow(Schema schema, RowTypeInfo typeInfo, IndexedRecord record) {
diff --git a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowFormatFactory.java b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowFormatFactory.java
new file mode 100644
index 0000000000000..1f46af2c782fa
--- /dev/null
+++ b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowFormatFactory.java
@@ -0,0 +1,97 @@
+/*
+ * 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.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.table.descriptors.AvroValidator;
+import org.apache.flink.table.descriptors.DescriptorProperties;
+import org.apache.flink.table.descriptors.FormatDescriptorValidator;
+import org.apache.flink.table.formats.DeserializationSchemaFactory;
+import org.apache.flink.table.formats.SerializationSchemaFactory;
+import org.apache.flink.types.Row;
+
+import org.apache.avro.specific.SpecificRecord;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Table format factory for providing configured instances of Avro-to-row {@link SerializationSchema}
+ * and {@link DeserializationSchema}.
+ */
+public class AvroRowFormatFactory implements SerializationSchemaFactory, DeserializationSchemaFactory {
+
+ @Override
+ public Map requiredContext() {
+ final Map context = new HashMap<>();
+ context.put(FormatDescriptorValidator.FORMAT_TYPE(), AvroValidator.FORMAT_TYPE_VALUE);
+ context.put(FormatDescriptorValidator.FORMAT_PROPERTY_VERSION(), "1");
+ return context;
+ }
+
+ @Override
+ public boolean supportsSchemaDerivation() {
+ return false;
+ }
+
+ @Override
+ public List supportedProperties() {
+ final List properties = new ArrayList<>();
+ properties.add(AvroValidator.FORMAT_RECORD_CLASS);
+ properties.add(AvroValidator.FORMAT_AVRO_SCHEMA);
+ return properties;
+ }
+
+ @Override
+ public DeserializationSchema createDeserializationSchema(Map properties) {
+ final DescriptorProperties descriptorProperties = new DescriptorProperties(true);
+ descriptorProperties.putProperties(properties);
+
+ // validate
+ new AvroValidator().validate(descriptorProperties);
+
+ // create and configure
+ if (descriptorProperties.containsKey(AvroValidator.FORMAT_RECORD_CLASS)) {
+ return new AvroRowDeserializationSchema(
+ descriptorProperties.getClass(AvroValidator.FORMAT_RECORD_CLASS, SpecificRecord.class));
+ } else {
+ return new AvroRowDeserializationSchema(descriptorProperties.getString(AvroValidator.FORMAT_AVRO_SCHEMA));
+ }
+ }
+
+ @Override
+ public SerializationSchema createSerializationSchema(Map properties) {
+ final DescriptorProperties descriptorProperties = new DescriptorProperties(true);
+ descriptorProperties.putProperties(properties);
+
+ // validate
+ new AvroValidator().validate(descriptorProperties);
+
+ // create and configure
+ if (descriptorProperties.containsKey(AvroValidator.FORMAT_RECORD_CLASS)) {
+ return new AvroRowSerializationSchema(
+ descriptorProperties.getClass(AvroValidator.FORMAT_RECORD_CLASS, SpecificRecord.class));
+ } else {
+ return new AvroRowSerializationSchema(descriptorProperties.getString(AvroValidator.FORMAT_AVRO_SCHEMA));
+ }
+ }
+}
diff --git a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowSerializationSchema.java b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowSerializationSchema.java
index 80f5f1d2da004..d4c731975d8f0 100644
--- a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowSerializationSchema.java
+++ b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowSerializationSchema.java
@@ -51,6 +51,7 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.TimeZone;
/**
@@ -147,6 +148,23 @@ public byte[] serialize(Row row) {
}
}
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ final AvroRowSerializationSchema that = (AvroRowSerializationSchema) o;
+ return Objects.equals(recordClazz, that.recordClazz) && Objects.equals(schemaString, that.schemaString);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(recordClazz, schemaString);
+ }
+
// --------------------------------------------------------------------------------------------
private GenericRecord convertRowToAvroRecord(Schema schema, Row row) {
diff --git a/flink-formats/flink-avro/src/main/resources/META-INF/services/org.apache.flink.table.formats.TableFormatFactory b/flink-formats/flink-avro/src/main/resources/META-INF/services/org.apache.flink.table.formats.TableFormatFactory
new file mode 100644
index 0000000000000..e24424a8dc6bd
--- /dev/null
+++ b/flink-formats/flink-avro/src/main/resources/META-INF/services/org.apache.flink.table.formats.TableFormatFactory
@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.flink.formats.avro.AvroRowFormatFactory
diff --git a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroRowFormatFactoryTest.java b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroRowFormatFactoryTest.java
new file mode 100644
index 0000000000000..4f5e21890bd41
--- /dev/null
+++ b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroRowFormatFactoryTest.java
@@ -0,0 +1,104 @@
+/*
+ * 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.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.formats.avro.generated.User;
+import org.apache.flink.table.descriptors.Avro;
+import org.apache.flink.table.descriptors.Descriptor;
+import org.apache.flink.table.descriptors.DescriptorProperties;
+import org.apache.flink.table.formats.DeserializationSchemaFactory;
+import org.apache.flink.table.formats.SerializationSchemaFactory;
+import org.apache.flink.table.formats.TableFormatFactoryService;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for the {@link AvroRowFormatFactory}.
+ */
+public class AvroRowFormatFactoryTest extends TestLogger {
+
+ private static final Class AVRO_SPECIFIC_RECORD = User.class;
+
+ private static final String AVRO_SCHEMA = User.getClassSchema().toString();
+
+ @Test
+ public void testRecordClass() {
+ final Map properties = toMap(new Avro().recordClass(AVRO_SPECIFIC_RECORD));
+
+ testRecordClassDeserializationSchema(properties);
+
+ testRecordClassSerializationSchema(properties);
+ }
+
+ @Test
+ public void testAvroSchema() {
+ final Map properties = toMap(new Avro().avroSchema(AVRO_SCHEMA));
+
+ testAvroSchemaSerializationSchema(properties);
+
+ testAvroSchemaDeserializationSchema(properties);
+ }
+
+ private void testRecordClassSerializationSchema(Map properties) {
+ final DeserializationSchema actual2 = TableFormatFactoryService
+ .find(DeserializationSchemaFactory.class, properties)
+ .createDeserializationSchema(properties);
+ final AvroRowDeserializationSchema expected2 = new AvroRowDeserializationSchema(AVRO_SPECIFIC_RECORD);
+ assertEquals(expected2, actual2);
+ }
+
+ private void testRecordClassDeserializationSchema(Map properties) {
+ final SerializationSchema actual1 = TableFormatFactoryService
+ .find(SerializationSchemaFactory.class, properties)
+ .createSerializationSchema(properties);
+ final SerializationSchema expected1 = new AvroRowSerializationSchema(AVRO_SPECIFIC_RECORD);
+ assertEquals(expected1, actual1);
+ }
+
+ private void testAvroSchemaDeserializationSchema(Map properties) {
+ final DeserializationSchema actual2 = TableFormatFactoryService
+ .find(DeserializationSchemaFactory.class, properties)
+ .createDeserializationSchema(properties);
+ final AvroRowDeserializationSchema expected2 = new AvroRowDeserializationSchema(AVRO_SCHEMA);
+ assertEquals(expected2, actual2);
+ }
+
+ private void testAvroSchemaSerializationSchema(Map properties) {
+ final SerializationSchema actual1 = TableFormatFactoryService
+ .find(SerializationSchemaFactory.class, properties)
+ .createSerializationSchema(properties);
+ final SerializationSchema expected1 = new AvroRowSerializationSchema(AVRO_SCHEMA);
+ assertEquals(expected1, actual1);
+ }
+
+ private static Map toMap(Descriptor... desc) {
+ final DescriptorProperties descriptorProperties = new DescriptorProperties(true);
+ for (Descriptor d : desc) {
+ d.addProperties(descriptorProperties);
+ }
+ return descriptorProperties.asMap();
+ }
+}
diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDeserializationSchema.java b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDeserializationSchema.java
index df5285190e55d..dc8a116ac62f2 100644
--- a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDeserializationSchema.java
+++ b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowDeserializationSchema.java
@@ -38,6 +38,7 @@
import java.sql.Date;
import java.sql.Time;
import java.sql.Timestamp;
+import java.util.Objects;
/**
* Deserialization schema from JSON to Flink types.
@@ -84,7 +85,7 @@ public JsonRowDeserializationSchema(TypeInformation typeInfo) {
* @see http://json-schema.org/
*/
public JsonRowDeserializationSchema(String jsonSchema) {
- this(JsonSchemaConverter.convert(jsonSchema));
+ this(JsonRowSchemaConverter.convert(jsonSchema));
}
@Override
@@ -118,6 +119,23 @@ public void setFailOnMissingField(boolean failOnMissingField) {
this.failOnMissingField = failOnMissingField;
}
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ final JsonRowDeserializationSchema that = (JsonRowDeserializationSchema) o;
+ return failOnMissingField == that.failOnMissingField && Objects.equals(typeInfo, that.typeInfo);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(typeInfo, failOnMissingField);
+ }
+
// --------------------------------------------------------------------------------------------
private Object convert(JsonNode node, TypeInformation info) {
diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowFormatFactory.java b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowFormatFactory.java
new file mode 100644
index 0000000000000..42e13a2b05327
--- /dev/null
+++ b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowFormatFactory.java
@@ -0,0 +1,106 @@
+/*
+ * 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.json;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.table.descriptors.DescriptorProperties;
+import org.apache.flink.table.descriptors.FormatDescriptorValidator;
+import org.apache.flink.table.descriptors.JsonValidator;
+import org.apache.flink.table.descriptors.SchemaValidator;
+import org.apache.flink.table.formats.DeserializationSchemaFactory;
+import org.apache.flink.table.formats.SerializationSchemaFactory;
+import org.apache.flink.types.Row;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Table format factory for providing configured instances of JSON-to-row {@link SerializationSchema}
+ * and {@link DeserializationSchema}.
+ */
+public class JsonRowFormatFactory implements SerializationSchemaFactory, DeserializationSchemaFactory {
+
+ @Override
+ public Map requiredContext() {
+ final Map context = new HashMap<>();
+ context.put(FormatDescriptorValidator.FORMAT_TYPE(), JsonValidator.FORMAT_TYPE_VALUE);
+ context.put(FormatDescriptorValidator.FORMAT_PROPERTY_VERSION(), "1");
+ return context;
+ }
+
+ @Override
+ public boolean supportsSchemaDerivation() {
+ return true;
+ }
+
+ @Override
+ public List supportedProperties() {
+ final List properties = new ArrayList<>();
+ properties.add(JsonValidator.FORMAT_JSON_SCHEMA);
+ properties.add(JsonValidator.FORMAT_SCHEMA);
+ properties.add(JsonValidator.FORMAT_FAIL_ON_MISSING_FIELD);
+ properties.add(FormatDescriptorValidator.FORMAT_DERIVE_SCHEMA());
+ properties.addAll(SchemaValidator.getSchemaDerivationKeys());
+ return properties;
+ }
+
+ @Override
+ public DeserializationSchema createDeserializationSchema(Map properties) {
+ final DescriptorProperties descriptorProperties = new DescriptorProperties(true);
+ descriptorProperties.putProperties(properties);
+
+ // validate
+ new JsonValidator().validate(descriptorProperties);
+
+ // create and configure
+ final JsonRowDeserializationSchema schema = new JsonRowDeserializationSchema(createTypeInformation(descriptorProperties));
+
+ descriptorProperties.getOptionalBoolean(JsonValidator.FORMAT_FAIL_ON_MISSING_FIELD)
+ .ifPresent(schema::setFailOnMissingField);
+
+ return schema;
+ }
+
+ @Override
+ public SerializationSchema createSerializationSchema(Map properties) {
+ final DescriptorProperties descriptorProperties = new DescriptorProperties(true);
+ descriptorProperties.putProperties(properties);
+
+ // validate
+ new JsonValidator().validate(descriptorProperties);
+
+ // create and configure
+ return new JsonRowSerializationSchema(createTypeInformation(descriptorProperties));
+ }
+
+ private TypeInformation createTypeInformation(DescriptorProperties descriptorProperties) {
+ if (descriptorProperties.containsKey(JsonValidator.FORMAT_SCHEMA)) {
+ return (RowTypeInfo) descriptorProperties.getType(JsonValidator.FORMAT_SCHEMA);
+ } else if (descriptorProperties.containsKey(JsonValidator.FORMAT_JSON_SCHEMA)) {
+ return JsonRowSchemaConverter.convert(descriptorProperties.getString(JsonValidator.FORMAT_JSON_SCHEMA));
+ } else {
+ return SchemaValidator.deriveFormatFields(descriptorProperties).toRowType();
+ }
+ }
+}
diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonSchemaConverter.java b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowSchemaConverter.java
similarity index 99%
rename from flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonSchemaConverter.java
rename to flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowSchemaConverter.java
index 7a001f63753e3..320ca1f8c4d5d 100644
--- a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonSchemaConverter.java
+++ b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowSchemaConverter.java
@@ -49,9 +49,9 @@
* This converter has been developed for JSON Schema draft-07 but also includes keywords of
* older drafts to be as compatible as possible.
*/
-public final class JsonSchemaConverter {
+public final class JsonRowSchemaConverter {
- private JsonSchemaConverter() {
+ private JsonRowSchemaConverter() {
// private
}
diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowSerializationSchema.java b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowSerializationSchema.java
index 8fee6a4d944e3..d9420629fea85 100644
--- a/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowSerializationSchema.java
+++ b/flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowSerializationSchema.java
@@ -39,6 +39,7 @@
import java.sql.Time;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
+import java.util.Objects;
/**
* Serialization schema that serializes an object of Flink types into a JSON bytes.
@@ -89,7 +90,7 @@ public JsonRowSerializationSchema(TypeInformation typeInfo) {
* @see http://json-schema.org/
*/
public JsonRowSerializationSchema(String jsonSchema) {
- this(JsonSchemaConverter.convert(jsonSchema));
+ this(JsonRowSchemaConverter.convert(jsonSchema));
}
@Override
@@ -107,6 +108,23 @@ public byte[] serialize(Row row) {
}
}
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ final JsonRowSerializationSchema that = (JsonRowSerializationSchema) o;
+ return Objects.equals(typeInfo, that.typeInfo);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(typeInfo);
+ }
+
// --------------------------------------------------------------------------------------------
private ObjectNode convertRow(ObjectNode reuse, RowTypeInfo info, Row row) {
diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/table/descriptors/Json.java b/flink-formats/flink-json/src/main/java/org/apache/flink/table/descriptors/Json.java
index 9c121916d5416..035f05f7cd540 100644
--- a/flink-formats/flink-json/src/main/java/org/apache/flink/table/descriptors/Json.java
+++ b/flink-formats/flink-json/src/main/java/org/apache/flink/table/descriptors/Json.java
@@ -20,6 +20,7 @@
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.table.typeutils.TypeStringUtils;
+import org.apache.flink.types.Row;
import org.apache.flink.util.Preconditions;
import static org.apache.flink.table.descriptors.FormatDescriptorValidator.FORMAT_DERIVE_SCHEMA;
@@ -81,7 +82,7 @@ public Json jsonSchema(String jsonSchema) {
*
* @param schemaType type information that describes the schema
*/
- public Json schema(TypeInformation schemaType) {
+ public Json schema(TypeInformation schemaType) {
Preconditions.checkNotNull(schemaType);
this.schema = TypeStringUtils.writeTypeInfo(schemaType);
this.jsonSchema = null;
diff --git a/flink-formats/flink-json/src/main/java/org/apache/flink/table/descriptors/JsonValidator.java b/flink-formats/flink-json/src/main/java/org/apache/flink/table/descriptors/JsonValidator.java
index fea7cf55b5d33..49e1abc8a1995 100644
--- a/flink-formats/flink-json/src/main/java/org/apache/flink/table/descriptors/JsonValidator.java
+++ b/flink-formats/flink-json/src/main/java/org/apache/flink/table/descriptors/JsonValidator.java
@@ -45,7 +45,7 @@ public void validate(DescriptorProperties properties) {
} else if (!deriveSchema && !hasSchema && !hasSchemaString) {
throw new ValidationException("A definition of a schema or JSON schema is required.");
} else if (hasSchema) {
- properties.validateType(FORMAT_SCHEMA, false);
+ properties.validateType(FORMAT_SCHEMA, true, false);
} else if (hasSchemaString) {
properties.validateString(FORMAT_JSON_SCHEMA, false, 1);
}
diff --git a/flink-formats/flink-json/src/main/resources/META-INF/services/org.apache.flink.table.formats.TableFormatFactory b/flink-formats/flink-json/src/main/resources/META-INF/services/org.apache.flink.table.formats.TableFormatFactory
new file mode 100644
index 0000000000000..aec584688eb02
--- /dev/null
+++ b/flink-formats/flink-json/src/main/resources/META-INF/services/org.apache.flink.table.formats.TableFormatFactory
@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.flink.formats.json.JsonRowFormatFactory
diff --git a/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowFormatFactoryTest.java b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowFormatFactoryTest.java
new file mode 100644
index 0000000000000..d763b9047139c
--- /dev/null
+++ b/flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowFormatFactoryTest.java
@@ -0,0 +1,150 @@
+/*
+ * 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.json;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.table.api.Types;
+import org.apache.flink.table.descriptors.Descriptor;
+import org.apache.flink.table.descriptors.DescriptorProperties;
+import org.apache.flink.table.descriptors.Json;
+import org.apache.flink.table.descriptors.Schema;
+import org.apache.flink.table.formats.DeserializationSchemaFactory;
+import org.apache.flink.table.formats.SerializationSchemaFactory;
+import org.apache.flink.table.formats.TableFormatFactoryService;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Test;
+
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for the {@link JsonRowFormatFactory}.
+ */
+public class JsonRowFormatFactoryTest extends TestLogger {
+
+ private static final String JSON_SCHEMA =
+ "{" +
+ " 'title': 'Fruit'," +
+ " 'type': 'object'," +
+ " 'properties': {" +
+ " 'name': {" +
+ " 'type': 'string'" +
+ " }," +
+ " 'count': {" +
+ " 'type': 'integer'" +
+ " }," +
+ " 'time': {" +
+ " 'description': 'row time'," +
+ " 'type': 'string'," +
+ " 'format': 'date-time'" +
+ " }" +
+ " }," +
+ " 'required': ['name', 'count', 'time']" +
+ "}";
+
+ private static final TypeInformation SCHEMA = Types.ROW(
+ new String[]{"field1", "field2"},
+ new TypeInformation[]{Types.BOOLEAN(), Types.INT()});
+
+ @Test
+ public void testSchema() {
+ final Map properties = toMap(
+ new Json()
+ .schema(SCHEMA)
+ .failOnMissingField(false));
+
+ testSchemaSerializationSchema(properties);
+
+ testSchemaDeserializationSchema(properties);
+ }
+
+ @Test
+ public void testJsonSchema() {
+ final Map properties = toMap(
+ new Json()
+ .jsonSchema(JSON_SCHEMA)
+ .failOnMissingField(true));
+
+ testJsonSchemaSerializationSchema(properties);
+
+ testJsonSchemaDeserializationSchema(properties);
+ }
+
+ @Test
+ public void testSchemaDerivation() {
+ final Map properties = toMap(
+ new Schema()
+ .field("field1", Types.BOOLEAN())
+ .field("field2", Types.INT())
+ .field("proctime", Types.SQL_TIMESTAMP()).proctime(),
+ new Json()
+ .deriveSchema());
+
+ testSchemaSerializationSchema(properties);
+
+ testSchemaDeserializationSchema(properties);
+ }
+
+ private void testSchemaDeserializationSchema(Map properties) {
+ final DeserializationSchema actual2 = TableFormatFactoryService
+ .find(DeserializationSchemaFactory.class, properties)
+ .createDeserializationSchema(properties);
+ final JsonRowDeserializationSchema expected2 = new JsonRowDeserializationSchema(SCHEMA);
+ expected2.setFailOnMissingField(false);
+ assertEquals(expected2, actual2);
+ }
+
+ private void testSchemaSerializationSchema(Map properties) {
+ final SerializationSchema actual1 = TableFormatFactoryService
+ .find(SerializationSchemaFactory.class, properties)
+ .createSerializationSchema(properties);
+ final SerializationSchema expected1 = new JsonRowSerializationSchema(SCHEMA);
+ assertEquals(expected1, actual1);
+ }
+
+ private void testJsonSchemaDeserializationSchema(Map properties) {
+ final DeserializationSchema actual2 = TableFormatFactoryService
+ .find(DeserializationSchemaFactory.class, properties)
+ .createDeserializationSchema(properties);
+ final JsonRowDeserializationSchema expected2 = new JsonRowDeserializationSchema(JSON_SCHEMA);
+ expected2.setFailOnMissingField(true);
+ assertEquals(expected2, actual2);
+ }
+
+ private void testJsonSchemaSerializationSchema(Map properties) {
+ final SerializationSchema actual1 = TableFormatFactoryService
+ .find(SerializationSchemaFactory.class, properties)
+ .createSerializationSchema(properties);
+ final SerializationSchema expected1 = new JsonRowSerializationSchema(JSON_SCHEMA);
+ assertEquals(expected1, actual1);
+ }
+
+ private static Map