[
https://issues.apache.org/jira/browse/FLINK-8558?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16534874#comment-16534874
]
ASF GitHub Bot commented on FLINK-8558:
---------------------------------------
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200651952
--- Diff:
flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceFactoryTestBase.java
---
@@ -0,0 +1,175 @@
+/*
+ * 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.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.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.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Abstract test base for {@link KafkaTableSourceFactory}.
+ */
+public abstract class KafkaTableSourceFactoryTestBase {
+
+ @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 String proctimeAttribute = "proc-time";
+
+ final List<RowtimeAttributeDescriptor>
rowtimeAttributeDescriptors = Collections.singletonList(
+ new RowtimeAttributeDescriptor("event-time", new
ExistingField("time"), new AscendingTimestamps()));
+
+ final Map<String, String> fieldMapping = new HashMap<>();
+ fieldMapping.put("fruit-name", "name");
+ fieldMapping.put("count", "count");
+
+ final String topic = "myTopic";
+
+ final Properties kafkaProperties = new Properties();
+ kafkaProperties.setProperty("zookeeper.connect", "dummy");
+ kafkaProperties.setProperty("group.id", "dummy");
+
+ final Map<KafkaTopicPartition, Long> specificOffsets = new
HashMap<>();
+ specificOffsets.put(new KafkaTopicPartition(topic, 0), 100L);
+ specificOffsets.put(new KafkaTopicPartition(topic, 1), 123L);
+
+ 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 = getKafkaTableSource(
+ schema,
+ proctimeAttribute,
+ rowtimeAttributeDescriptors,
+ fieldMapping,
+ topic,
+ kafkaProperties,
+ deserializationSchema,
+ startupMode,
+ specificOffsets);
+
+ // construct table source using descriptors and table source
factory
+
+ final Map<Integer, Long> offsets = new HashMap<>();
+ offsets.put(0, 100L);
+ offsets.put(1, 123L);
+
+ final TestTableSourceDescriptor testDesc = new
TestTableSourceDescriptor(
+ new Kafka()
+ .version(getKafkaVersion())
+ .topic(topic)
+ .properties(kafkaProperties)
+ .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<?> foundSource =
TableSourceFactoryService.findAndCreateTableSource(testDesc);
+
+ assertEquals(expected, foundSource);
+
+ // test Kafka consumer
+ final KafkaTableSource kafkaSource = (KafkaTableSource)
foundSource;
+ final KafkaTableSource observed = spy(kafkaSource);
+ final StreamExecutionEnvironment env =
mock(StreamExecutionEnvironment.class);
+
when(env.addSource(any(SourceFunction.class))).thenReturn(mock(DataStreamSource.class));
+ observed.getDataStream(env);
+
+ verify(env).addSource(any(getFlinkKafkaConsumer()));
+
+ verify(observed).getKafkaConsumer(
+ eq(topic),
+ // the consumer modifies the properties that's why we
compare them this way
+ eq(kafkaSource.getProperties()),
+ eq(deserializationSchema));
+ }
+
+ //
--------------------------------------------------------------------------------------------
+ // For version-specific tests
+ //
--------------------------------------------------------------------------------------------
+
+ protected abstract String getKafkaVersion();
+
+ protected abstract Class<FlinkKafkaConsumerBase<Row>>
getFlinkKafkaConsumer();
+
+ protected abstract KafkaTableSource getKafkaTableSource(
--- End diff --
`getExpectedKafkaTableSource`
> Add unified format interfaces and format discovery
> --------------------------------------------------
>
> Key: FLINK-8558
> URL: https://issues.apache.org/jira/browse/FLINK-8558
> Project: Flink
> Issue Type: New Feature
> Components: Streaming Connectors
> Reporter: Timo Walther
> Assignee: Timo Walther
> Priority: Major
> Labels: pull-request-available
>
> In the last release, we introduced a new module {{flink-formats}}. Currently
> only {{flink-avro}} is located there but we will add more formats such as
> {{flink-json}}, {{flink-protobuf}}, and so on. For better separation of
> concerns we want decouple connectors from formats: e.g., remove
> {{KafkaAvroTableSource}} and {{KafkaJsonTableSource}}.
> A newly introduced {{FormatFactory}} will use Java service loaders to
> discovery available formats in the classpath (similar to how file systems are
> discovered now). A {{Format}} will provide a method for converting {{byte[]}}
> to target record type.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
