[FLINK-8538] [table] Improve unified table sources

This closes apache#5564.
laubersder · Feb 27, 2018 · db2c510 · db2c510
1 parent 1d26062
commit db2c510
Show file tree

Hide file tree

Showing 71 changed files with 2,561 additions and 1,451 deletions.
diff --git a/flink-connectors/flink-connector-kafka-0.10/pom.xml b/flink-connectors/flink-connector-kafka-0.10/pom.xml
@@ -192,6 +192,14 @@ under the License.
  <scope>test</scope>
  </dependency>
 
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
  </dependencies>
 
  <build>

diff --git a/...main/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSourceFactory.java b/...main/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSourceFactory.java
@@ -18,19 +18,20 @@
 
 package org.apache.flink.streaming.connectors.kafka;
 
-import static org.apache.flink.table.descriptors.KafkaValidator.KAFKA_VERSION_VALUE_010;
+import static org.apache.flink.table.descriptors.KafkaValidator.CONNECTOR_VERSION_VALUE_010;
 
 /**
  * Factory for creating configured instances of {@link Kafka010JsonTableSource}.
  */
 public class Kafka010JsonTableSourceFactory extends KafkaJsonTableSourceFactory {
+
  @Override
  protected KafkaJsonTableSource.Builder createBuilder() {
  return new Kafka010JsonTableSource.Builder();
  }
 
  @Override
  protected String kafkaVersion() {
- return KAFKA_VERSION_VALUE_010;
+ return CONNECTOR_VERSION_VALUE_010;
  }
 }
diff --git a/...-connectors/flink-connector-kafka-0.10/src/main/resources/tableSourceConverter.properties b/...-connectors/flink-connector-kafka-0.10/src/main/resources/tableSourceConverter.properties
diff --git a/...kafka/Kafka010TableSourceFactoryTest.java → ...a/Kafka010JsonTableSourceFactoryTest.java b/...kafka/Kafka010TableSourceFactoryTest.java → ...a/Kafka010JsonTableSourceFactoryTest.java
@@ -18,24 +18,20 @@
 
 package org.apache.flink.streaming.connectors.kafka;
 
-import org.apache.flink.table.descriptors.Kafka;
-
-import static org.apache.flink.table.descriptors.KafkaValidator.KAFKA_VERSION_VALUE_010;
+import static org.apache.flink.table.descriptors.KafkaValidator.CONNECTOR_VERSION_VALUE_010;
 
 /**
  * Tests for {@link Kafka010JsonTableSourceFactory}.
  */
-public class Kafka010TableSourceFactoryTest extends KafkaJsonTableFromDescriptorTestBase {
- protected String versionForTest() {
- return KAFKA_VERSION_VALUE_010;
- }
+public class Kafka010JsonTableSourceFactoryTest extends KafkaJsonTableSourceFactoryTestBase {
 
- protected KafkaJsonTableSource.Builder builderForTest() {
- return Kafka010JsonTableSource.builder();
+ @Override
+ protected String version() {
+ return CONNECTOR_VERSION_VALUE_010;
  }
 
  @Override
- protected void extraSettings(KafkaTableSource.Builder builder, Kafka kafka) {
- // no extra settings
+ protected KafkaJsonTableSource.Builder builder() {
+ return Kafka010JsonTableSource.builder();
  }
 }
diff --git a/flink-connectors/flink-connector-kafka-0.11/pom.xml b/flink-connectors/flink-connector-kafka-0.11/pom.xml
@@ -201,6 +201,14 @@ under the License.
  <scope>test</scope>
  </dependency>
 
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
  </dependencies>
 
  <build>

diff --git a/...main/java/org/apache/flink/streaming/connectors/kafka/Kafka011JsonTableSourceFactory.java b/...main/java/org/apache/flink/streaming/connectors/kafka/Kafka011JsonTableSourceFactory.java
@@ -18,19 +18,20 @@
 
 package org.apache.flink.streaming.connectors.kafka;
 
-import static org.apache.flink.table.descriptors.KafkaValidator.KAFKA_VERSION_VALUE_011;
+import static org.apache.flink.table.descriptors.KafkaValidator.CONNECTOR_VERSION_VALUE_011;
 
 /**
  * Factory for creating configured instances of {@link Kafka011JsonTableSource}.
  */
 public class Kafka011JsonTableSourceFactory extends KafkaJsonTableSourceFactory {
+
  @Override
  protected KafkaJsonTableSource.Builder createBuilder() {
  return new Kafka011JsonTableSource.Builder();
  }
 
  @Override
  protected String kafkaVersion() {
- return KAFKA_VERSION_VALUE_011;
+ return CONNECTOR_VERSION_VALUE_011;
  }
 }
diff --git a/...-connectors/flink-connector-kafka-0.11/src/main/resources/tableSourceConverter.properties b/...-connectors/flink-connector-kafka-0.11/src/main/resources/tableSourceConverter.properties
diff --git a/...kafka/Kafka011TableSourceFactoryTest.java → ...a/Kafka011JsonTableSourceFactoryTest.java b/...kafka/Kafka011TableSourceFactoryTest.java → ...a/Kafka011JsonTableSourceFactoryTest.java
@@ -18,24 +18,20 @@
 
 package org.apache.flink.streaming.connectors.kafka;
 
-import org.apache.flink.table.descriptors.Kafka;
-
-import static org.apache.flink.table.descriptors.KafkaValidator.KAFKA_VERSION_VALUE_011;
+import static org.apache.flink.table.descriptors.KafkaValidator.CONNECTOR_VERSION_VALUE_011;
 
 /**
  * Tests for {@link Kafka011JsonTableSourceFactory}.
  */
-public class Kafka011TableSourceFactoryTest extends KafkaJsonTableFromDescriptorTestBase {
- protected String versionForTest() {
- return KAFKA_VERSION_VALUE_011;
- }
+public class Kafka011JsonTableSourceFactoryTest extends KafkaJsonTableSourceFactoryTestBase {
 
- protected KafkaJsonTableSource.Builder builderForTest() {
- return Kafka011JsonTableSource.builder();
+ @Override
+ protected String version() {
+ return CONNECTOR_VERSION_VALUE_011;
  }
 
  @Override
- protected void extraSettings(KafkaTableSource.Builder builder, Kafka kafka) {
- // no extra settings
+ protected KafkaJsonTableSource.Builder builder() {
+ return Kafka011JsonTableSource.builder();
  }
 }
diff --git a/flink-connectors/flink-connector-kafka-0.8/pom.xml b/flink-connectors/flink-connector-kafka-0.8/pom.xml
@@ -204,6 +204,14 @@ under the License.
  <scope>test</scope>
  </dependency>
 
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
  </dependencies>
 
  <build>

diff --git a/.../main/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSourceFactory.java b/.../main/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSourceFactory.java
@@ -18,19 +18,20 @@
 
 package org.apache.flink.streaming.connectors.kafka;
 
-import static org.apache.flink.table.descriptors.KafkaValidator.KAFKA_VERSION_VALUE_08;
+import static org.apache.flink.table.descriptors.KafkaValidator.CONNECTOR_VERSION_VALUE_08;
 
 /**
  * Factory for creating configured instances of {@link Kafka08JsonTableSource}.
  */
 public class Kafka08JsonTableSourceFactory extends KafkaJsonTableSourceFactory {
+
  @Override
  protected KafkaJsonTableSource.Builder createBuilder() {
  return new Kafka08JsonTableSource.Builder();
  }
 
  @Override
  protected String kafkaVersion() {
- return KAFKA_VERSION_VALUE_08;
+ return CONNECTOR_VERSION_VALUE_08;
  }
 }
diff --git a/...k-connectors/flink-connector-kafka-0.8/src/main/resources/tableSourceConverter.properties b/...k-connectors/flink-connector-kafka-0.8/src/main/resources/tableSourceConverter.properties
diff --git a/.../kafka/Kafka08TableSourceFactoryTest.java → ...ka/Kafka08JsonTableSourceFactoryTest.java b/.../kafka/Kafka08TableSourceFactoryTest.java → ...ka/Kafka08JsonTableSourceFactoryTest.java
@@ -18,25 +18,20 @@
 
 package org.apache.flink.streaming.connectors.kafka;
 
-import org.apache.flink.table.descriptors.Kafka;
-
-import static org.apache.flink.table.descriptors.KafkaValidator.KAFKA_VERSION_VALUE_08;
+import static org.apache.flink.table.descriptors.KafkaValidator.CONNECTOR_VERSION_VALUE_08;
 
 /**
  * Tests for {@link Kafka08JsonTableSourceFactory}.
  */
-public class Kafka08TableSourceFactoryTest extends KafkaJsonTableFromDescriptorTestBase {
- protected String versionForTest() {
- return KAFKA_VERSION_VALUE_08;
- }
+public class Kafka08JsonTableSourceFactoryTest extends KafkaJsonTableSourceFactoryTestBase {
 
- protected KafkaJsonTableSource.Builder builderForTest() {
- return Kafka08JsonTableSource.builder();
+ @Override
+ protected String version() {
+ return CONNECTOR_VERSION_VALUE_08;
  }
 
  @Override
- protected void extraSettings(KafkaTableSource.Builder builder, Kafka kafka) {
- builder.getKafkaProps().put("zookeeper.connect", "localhost:1111");
- kafka.zookeeperConnect("localhost:1111");
+ protected KafkaJsonTableSource.Builder builder() {
+ return Kafka08JsonTableSource.builder();
  }
 }
diff --git a/flink-connectors/flink-connector-kafka-0.9/pom.xml b/flink-connectors/flink-connector-kafka-0.9/pom.xml
@@ -173,6 +173,14 @@ under the License.
  <scope>test</scope>
  </dependency>
 
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
  <dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>hadoop-minikdc</artifactId>

diff --git a/.../main/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSourceFactory.java b/.../main/java/org/apache/flink/streaming/connectors/kafka/Kafka09JsonTableSourceFactory.java
@@ -18,19 +18,20 @@
 
 package org.apache.flink.streaming.connectors.kafka;
 
-import static org.apache.flink.table.descriptors.KafkaValidator.KAFKA_VERSION_VALUE_09;
+import static org.apache.flink.table.descriptors.KafkaValidator.CONNECTOR_VERSION_VALUE_09;
 
 /**
  * Factory for creating configured instances of {@link Kafka09JsonTableSource}.
  */
 public class Kafka09JsonTableSourceFactory extends KafkaJsonTableSourceFactory {
+
  @Override
  protected KafkaJsonTableSource.Builder createBuilder() {
  return new Kafka09JsonTableSource.Builder();
  }
 
  @Override
  protected String kafkaVersion() {
- return KAFKA_VERSION_VALUE_09;
+ return CONNECTOR_VERSION_VALUE_09;
  }
 }
diff --git a/...k-connectors/flink-connector-kafka-0.9/src/main/resources/tableSourceConverter.properties b/...k-connectors/flink-connector-kafka-0.9/src/main/resources/tableSourceConverter.properties