[FLINK-16125][connecotr/kafka] Remove Kafka connector property zookee…

…per.connect and clear documentation because Kafka 0.8 connector has been removed.

docs/dev/table/connect.md

@@ -159,7 +159,6 @@ CREATE TABLE MyUserTable (
  'connector.version' = '0.10',
  'connector.topic' = 'topic_name',
  'connector.startup-mode' = 'earliest-offset',
- 'connector.properties.zookeeper.connect' = 'localhost:2181',
  'connector.properties.bootstrap.servers' = 'localhost:9092',
 
  -- declare a format for this system
@@ -177,7 +176,6 @@ tableEnvironment
  .version("0.10")
  .topic("test-input")
  .startFromEarliest()
- .property("zookeeper.connect", "localhost:2181")
  .property("bootstrap.servers", "localhost:9092")
  )
 
@@ -211,7 +209,6 @@ table_environment \
  .version("0.10")
  .topic("test-input")
  .start_from_earliest()
- .property("zookeeper.connect", "localhost:2181")
  .property("bootstrap.servers", "localhost:9092")
  ) \
  .with_format( # declare a format for this system
@@ -246,7 +243,6 @@ tables:
  topic: test-input
  startup-mode: earliest-offset
  properties:
- zookeeper.connect: localhost:2181
  bootstrap.servers: localhost:9092
 
   # declare a format for this system
@@ -773,8 +769,6 @@ CREATE TABLE MyUserTable (
 
  'connector.topic' = 'topic_name', -- required: topic name from which the table is read
 
- -- required: specify the ZooKeeper connection string
- 'connector.properties.zookeeper.connect' = 'localhost:2181',
  -- required: specify the Kafka server connection string
  'connector.properties.bootstrap.servers' = 'localhost:9092',
  -- required for Kafka source, optional for Kafka sink, specify consumer group
@@ -814,7 +808,6 @@ CREATE TABLE MyUserTable (
  .topic("...") // required: topic name from which the table is read
 
  // optional: connector specific properties
- .property("zookeeper.connect", "localhost:2181")
  .property("bootstrap.servers", "localhost:9092")
  .property("group.id", "testGroup")
 
@@ -844,7 +837,6 @@ CREATE TABLE MyUserTable (
  .topic("...") # required: topic name from which the table is read
 
  # optional: connector specific properties
- .property("zookeeper.connect", "localhost:2181")
  .property("bootstrap.servers", "localhost:9092")
  .property("group.id", "testGroup")
 
@@ -874,7 +866,6 @@ connector:
  topic: ... # required: topic name from which the table is read
 
  properties:
- zookeeper.connect: localhost:2181 # required: specify the ZooKeeper connection string
  bootstrap.servers: localhost:9092 # required: specify the Kafka server connection string
  group.id: testGroup # optional: required in Kafka consumer, specify consumer group
 

docs/dev/table/connect.zh.md

@@ -159,7 +159,6 @@ CREATE TABLE MyUserTable (
  'connector.version' = '0.10',
  'connector.topic' = 'topic_name',
  'connector.startup-mode' = 'earliest-offset',
- 'connector.properties.zookeeper.connect' = 'localhost:2181',
  'connector.properties.bootstrap.servers' = 'localhost:9092',
 
  -- declare a format for this system
@@ -177,7 +176,6 @@ tableEnvironment
  .version("0.10")
  .topic("test-input")
  .startFromEarliest()
- .property("zookeeper.connect", "localhost:2181")
  .property("bootstrap.servers", "localhost:9092")
  )
 
@@ -211,7 +209,6 @@ table_environment \
  .version("0.10")
  .topic("test-input")
  .start_from_earliest()
- .property("zookeeper.connect", "localhost:2181")
  .property("bootstrap.servers", "localhost:9092")
  ) \
  .with_format( # declare a format for this system
@@ -246,7 +243,6 @@ tables:
  topic: test-input
  startup-mode: earliest-offset
  properties:
- zookeeper.connect: localhost:2181
  bootstrap.servers: localhost:9092
 
   # declare a format for this system
@@ -773,8 +769,6 @@ CREATE TABLE MyUserTable (
 
  'connector.topic' = 'topic_name', -- required: topic name from which the table is read
 
- -- required: specify the ZooKeeper connection string
- 'connector.properties.zookeeper.connect' = 'localhost:2181',
  -- required: specify the Kafka server connection string
  'connector.properties.bootstrap.servers' = 'localhost:9092',
  -- required for Kafka source, optional for Kafka sink, specify consumer group
@@ -814,7 +808,6 @@ CREATE TABLE MyUserTable (
  .topic("...") // required: topic name from which the table is read
 
  // optional: connector specific properties
- .property("zookeeper.connect", "localhost:2181")
  .property("bootstrap.servers", "localhost:9092")
  .property("group.id", "testGroup")
 
@@ -844,7 +837,6 @@ CREATE TABLE MyUserTable (
  .topic("...") # required: topic name from which the table is read
 
  # optional: connector specific properties
- .property("zookeeper.connect", "localhost:2181")
  .property("bootstrap.servers", "localhost:9092")
  .property("group.id", "testGroup")
 
@@ -874,7 +866,6 @@ connector:
  topic: ... # required: topic name from which the table is read
 
  properties:
- zookeeper.connect: localhost:2181 # required: specify the ZooKeeper connection string
  bootstrap.servers: localhost:9092 # required: specify the Kafka server connection string
  group.id: testGroup # optional: required in Kafka consumer, specify consumer group
 

docs/dev/table/hive/hive_catalog.md

@@ -190,7 +190,6 @@ Flink SQL> CREATE TABLE mykafka (name String, age Int) WITH (
  'connector.type' = 'kafka',
  'connector.version' = 'universal',
  'connector.topic' = 'test',
- 'connector.properties.zookeeper.connect' = 'localhost:2181',
  'connector.properties.bootstrap.servers' = 'localhost:9092',
  'format.type' = 'csv',
  'update-mode' = 'append'
@@ -227,7 +226,6 @@ Location: ......
 Table Type: MANAGED_TABLE
 Table Parameters:
  flink.connector.properties.bootstrap.servers localhost:9092
- flink.connector.properties.zookeeper.connect localhost:2181
  flink.connector.topic test
  flink.connector.type kafka
  flink.connector.version universal

docs/dev/table/hive/hive_catalog.zh.md

@@ -190,7 +190,6 @@ Flink SQL> CREATE TABLE mykafka (name String, age Int) WITH (
  'connector.type' = 'kafka',
  'connector.version' = 'universal',
  'connector.topic' = 'test',
- 'connector.properties.zookeeper.connect' = 'localhost:2181',
  'connector.properties.bootstrap.servers' = 'localhost:9092',
  'format.type' = 'csv',
  'update-mode' = 'append'
@@ -227,7 +226,6 @@ Location: ......
 Table Type: MANAGED_TABLE
 Table Parameters:
  flink.connector.properties.bootstrap.servers localhost:9092
- flink.connector.properties.zookeeper.connect localhost:2181
  flink.connector.topic test
  flink.connector.type kafka
  flink.connector.version universal

docs/dev/table/sqlClient.md

@@ -317,7 +317,6 @@ tables:
  topic: TaxiRides
  startup-mode: earliest-offset
  properties:
- zookeeper.connect: localhost:2181
  bootstrap.servers: localhost:9092
  group.id: testGroup
  format:
@@ -483,7 +482,6 @@ tables:
  version: "0.11"
  topic: OutputTopic
  properties:
- zookeeper.connect: localhost:2181
  bootstrap.servers: localhost:9092
  group.id: testGroup
  format:

docs/dev/table/sqlClient.zh.md

@@ -317,7 +317,6 @@ tables:
  topic: TaxiRides
  startup-mode: earliest-offset
  properties:
- zookeeper.connect: localhost:2181
  bootstrap.servers: localhost:9092
  group.id: testGroup
  format:
@@ -483,7 +482,6 @@ tables:
  version: "0.11"
  topic: OutputTopic
  properties:
- zookeeper.connect: localhost:2181
  bootstrap.servers: localhost:9092
  group.id: testGroup
  format:

flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java

@@ -253,7 +253,6 @@ public void prepare(Config config) throws Exception {
  LOG.info("ZK and KafkaServer started.");
 
  standardProps = new Properties();
- standardProps.setProperty("zookeeper.connect", zookeeperConnectionString);
  standardProps.setProperty("bootstrap.servers", brokerConnectionString);
  standardProps.setProperty("group.id", "flink-tests");
  standardProps.setProperty("enable.auto.commit", "false");

flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java

@@ -134,7 +134,6 @@ public void prepare(Config config) throws Exception {
  LOG.info("ZK and KafkaServer started.");
 
  standardProps = new Properties();
- standardProps.setProperty("zookeeper.connect", zookeeperConnectionString);
  standardProps.setProperty("bootstrap.servers", brokerConnectionString);
  standardProps.setProperty("group.id", "flink-tests");
  standardProps.setProperty("enable.auto.commit", "false");
@@ -393,8 +392,8 @@ protected KafkaServer getKafkaServer(int brokerId, File tmpFolder) throws Except
  kafkaProperties.put("advertised.host.name", KAFKA_HOST);
  kafkaProperties.put("broker.id", Integer.toString(brokerId));
  kafkaProperties.put("log.dir", tmpFolder.toString());
- kafkaProperties.put("zookeeper.connect", zookeeperConnectionString);
  kafkaProperties.put("message.max.bytes", String.valueOf(50 * 1024 * 1024));
+ kafkaProperties.put("zookeeper.connect", zookeeperConnectionString);
  kafkaProperties.put("replica.fetch.max.bytes", String.valueOf(50 * 1024 * 1024));
  kafkaProperties.put("transaction.max.timeout.ms", Integer.toString(1000 * 60 * 60 * 2)); // 2hours
 

flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/table/descriptors/KafkaValidator.java

@@ -54,7 +54,6 @@ public class KafkaValidator extends ConnectorDescriptorValidator {
  public static final String CONNECTOR_SPECIFIC_OFFSETS_OFFSET = "offset";
  public static final String CONNECTOR_STARTUP_TIMESTAMP_MILLIS = "connector.startup-timestamp-millis";
  public static final String CONNECTOR_PROPERTIES = "connector.properties";
- public static final String CONNECTOR_PROPERTIES_ZOOKEEPER_CONNECT = "connector.properties.zookeeper.connect";
  public static final String CONNECTOR_PROPERTIES_BOOTSTRAP_SERVER = "connector.properties.bootstrap.servers";
  public static final String CONNECTOR_PROPERTIES_GROUP_ID = "connector.properties.group.id";
  public static final String CONNECTOR_PROPERTIES_KEY = "key";
@@ -136,11 +135,9 @@ private void validateStartupMode(DescriptorProperties properties) {
  }
 
  private void validateKafkaProperties(DescriptorProperties properties) {
- if (properties.containsKey(CONNECTOR_PROPERTIES_ZOOKEEPER_CONNECT)
- || properties.containsKey(CONNECTOR_PROPERTIES_BOOTSTRAP_SERVER)
+ if (properties.containsKey(CONNECTOR_PROPERTIES_BOOTSTRAP_SERVER)
  || properties.containsKey(CONNECTOR_PROPERTIES_GROUP_ID)) {
 
- properties.validateString(CONNECTOR_PROPERTIES_ZOOKEEPER_CONNECT, false);
  properties.validateString(CONNECTOR_PROPERTIES_BOOTSTRAP_SERVER, false);
  properties.validateString(CONNECTOR_PROPERTIES_GROUP_ID, true);
 
@@ -235,8 +232,7 @@ public static Map<Integer, Long> validateAndParseSpecificOffsetsString(Descripto
  }
 
  public static boolean hasConciseKafkaProperties(DescriptorProperties descriptorProperties) {
- return descriptorProperties.containsKey(CONNECTOR_PROPERTIES_ZOOKEEPER_CONNECT) ||
- descriptorProperties.containsKey(CONNECTOR_PROPERTIES_BOOTSTRAP_SERVER) ||
+ return descriptorProperties.containsKey(CONNECTOR_PROPERTIES_BOOTSTRAP_SERVER) ||
  descriptorProperties.containsKey(CONNECTOR_PROPERTIES_GROUP_ID);
  }
 }

flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java

@@ -169,7 +169,6 @@ public void runFailOnNoBrokerTest() throws Exception {
 
  // use wrong ports for the consumers
  properties.setProperty("bootstrap.servers", "localhost:80");
- properties.setProperty("zookeeper.connect", "localhost:80");
  properties.setProperty("group.id", "test");
  properties.setProperty("request.timeout.ms", "3000"); // let the test fail fast
  properties.setProperty("socket.timeout.ms", "3000");

flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactoryTestBase.java

@@ -93,7 +93,6 @@ public abstract class KafkaTableSourceSinkFactoryTestBase extends TestLogger {
 
  private static final Properties KAFKA_PROPERTIES = new Properties();
  static {
- KAFKA_PROPERTIES.setProperty("zookeeper.connect", "dummy");
  KAFKA_PROPERTIES.setProperty("group.id", "dummy");
  KAFKA_PROPERTIES.setProperty("bootstrap.servers", "dummy");
  }
@@ -224,7 +223,6 @@ public void testTableSourceWithLegacyProperties() {
 
  // use legacy properties
  legacyPropertiesMap.remove("connector.specific-offsets");
- legacyPropertiesMap.remove("connector.properties.zookeeper.connect");
  legacyPropertiesMap.remove("connector.properties.bootstrap.servers");
  legacyPropertiesMap.remove("connector.properties.group.id");
 
@@ -236,12 +234,10 @@ public void testTableSourceWithLegacyProperties() {
  legacyPropertiesMap.put("connector.specific-offsets.0.offset", "100");
  legacyPropertiesMap.put("connector.specific-offsets.1.partition", "1");
  legacyPropertiesMap.put("connector.specific-offsets.1.offset", "123");
- legacyPropertiesMap.put("connector.properties.0.key", "zookeeper.connect");
+ legacyPropertiesMap.put("connector.properties.0.key", "bootstrap.servers");
  legacyPropertiesMap.put("connector.properties.0.value", "dummy");
- legacyPropertiesMap.put("connector.properties.1.key", "bootstrap.servers");
+ legacyPropertiesMap.put("connector.properties.1.key", "group.id");
  legacyPropertiesMap.put("connector.properties.1.value", "dummy");
- legacyPropertiesMap.put("connector.properties.2.key", "group.id");
- legacyPropertiesMap.put("connector.properties.2.value", "dummy");
 
  final TableSource<?> actualSource = TableFactoryService.find(StreamTableSourceFactory.class, legacyPropertiesMap)
  .createStreamTableSource(legacyPropertiesMap);
@@ -330,7 +326,6 @@ public void testTableSinkWithLegacyProperties() {
 
  // use legacy properties
  legacyPropertiesMap.remove("connector.specific-offsets");
- legacyPropertiesMap.remove("connector.properties.zookeeper.connect");
  legacyPropertiesMap.remove("connector.properties.bootstrap.servers");
  legacyPropertiesMap.remove("connector.properties.group.id");
 
@@ -342,12 +337,10 @@ public void testTableSinkWithLegacyProperties() {
  legacyPropertiesMap.put("connector.specific-offsets.0.offset", "100");
  legacyPropertiesMap.put("connector.specific-offsets.1.partition", "1");
  legacyPropertiesMap.put("connector.specific-offsets.1.offset", "123");
- legacyPropertiesMap.put("connector.properties.0.key", "zookeeper.connect");
+ legacyPropertiesMap.put("connector.properties.0.key", "bootstrap.servers");
  legacyPropertiesMap.put("connector.properties.0.value", "dummy");
- legacyPropertiesMap.put("connector.properties.1.key", "bootstrap.servers");
+ legacyPropertiesMap.put("connector.properties.1.key", "group.id");
  legacyPropertiesMap.put("connector.properties.1.value", "dummy");
- legacyPropertiesMap.put("connector.properties.2.key", "group.id");
- legacyPropertiesMap.put("connector.properties.2.value", "dummy");
 
  final TableSink<?> actualSink = TableFactoryService.find(StreamTableSinkFactory.class, legacyPropertiesMap)
  .createStreamTableSink(legacyPropertiesMap);

flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableTestBase.java

@@ -66,7 +66,6 @@ public void testKafkaSourceSink() throws Exception {
 
  // ---------- Produce an event time stream into Kafka -------------------
  String groupId = standardProps.getProperty("group.id");
- String zk = standardProps.getProperty("zookeeper.connect");
  String bootstraps = standardProps.getProperty("bootstrap.servers");
 
  // TODO: use DDL to register Kafka once FLINK-15282 is fixed.
@@ -83,7 +82,6 @@ public void testKafkaSourceSink() throws Exception {
  properties.put("connector.type", "kafka");
  properties.put("connector.topic", topic);
  properties.put("connector.version", kafkaVersion());
- properties.put("connector.properties.zookeeper.connect", zk);
  properties.put("connector.properties.bootstrap.servers", bootstraps);
  properties.put("connector.properties.group.id", groupId);
  properties.put("connector.startup-mode", "earliest-offset");
@@ -112,7 +110,6 @@ public void testKafkaSourceSink() throws Exception {
 // " 'connector.type' = 'kafka',\n" +
 // " 'connector.topic' = '" + topic + "',\n" +
 // " 'connector.version' = 'universal',\n" +
-// " 'connector.properties.zookeeper.connect' = '" + zk + "',\n" +
 // " 'connector.properties.bootstrap.servers' = '" + bootstraps + "',\n" +
 // " 'connector.properties.group.id' = '" + groupId + "', \n" +
 // " 'connector.startup-mode' = 'earliest-offset', \n" +

flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java

@@ -137,7 +137,6 @@ public void prepare(Config config) throws Exception {
  LOG.info("ZK and KafkaServer started.");
 
  standardProps = new Properties();
- standardProps.setProperty("zookeeper.connect", zookeeperConnectionString);
  standardProps.setProperty("bootstrap.servers", brokerConnectionString);
  standardProps.setProperty("group.id", "flink-tests");
  standardProps.setProperty("enable.auto.commit", "false");

flink-end-to-end-tests/flink-confluent-schema-registry/src/main/java/org/apache/flink/schema/registry/test/TestAvroConsumerConfluent.java

@@ -37,7 +37,7 @@
  * A simple example that shows how to read from and write to Kafka with Confluent Schema Registry.
  * This will read AVRO messages from the input topic, parse them into a POJO type via checking the Schema by calling Schema registry.
  * Then this example publish the POJO type to kafka by converting the POJO to AVRO and verifying the schema.
- * --input-topic test-input --output-string-topic test-output --output-avro-topic test-avro-output --output-subject --bootstrap.servers localhost:9092 --zookeeper.connect localhost:2181 --schema-registry-url http:https://localhost:8081 --group.id myconsumer
+ * --input-topic test-input --output-string-topic test-output --output-avro-topic test-avro-output --output-subject --bootstrap.servers localhost:9092 --schema-registry-url http:https://localhost:8081 --group.id myconsumer
  */
 public class TestAvroConsumerConfluent {
 
@@ -49,14 +49,12 @@ public static void main(String[] args) throws Exception {
  System.out.println("Missing parameters!\n" +
  "Usage: Kafka --input-topic <topic> --output-string-topic <topic> --output-avro-topic <topic> " +
  "--bootstrap.servers <kafka brokers> " +
- "--zookeeper.connect <zk quorum> " +
  "--schema-registry-url <confluent schema registry> --group.id <some id>");
  return;
  }
  Properties config = new Properties();
  config.setProperty("bootstrap.servers", parameterTool.getRequired("bootstrap.servers"));
  config.setProperty("group.id", parameterTool.getRequired("group.id"));
- config.setProperty("zookeeper.connect", parameterTool.getRequired("zookeeper.connect"));
  String schemaRegistryUrl = parameterTool.getRequired("schema-registry-url");
 
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/StreamingKafkaITCase.java

@@ -101,7 +101,6 @@ public void testKafka() throws Exception {
  .addArgument("--output-topic", outputTopic)
  .addArgument("--prefix", "PREFIX")
  .addArgument("--bootstrap.servers", kafka.getBootstrapServerAddresses().stream().map(address -> address.getHostString() + ':' + address.getPort()).collect(Collectors.joining(",")))
- .addArgument("--zookeeper.connect ", kafka.getZookeeperAddress().getHostString() + ':' + kafka.getZookeeperAddress().getPort())
  .addArgument("--group.id", "myconsumer")
  .addArgument("--auto.offset.reset", "earliest")
  .addArgument("--transaction.timeout.ms", "900000")

flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/src/test/resources/kafka_json_source_schema.yaml

@@ -40,7 +40,6 @@ tables:
  topic: $TOPIC_NAME
  startup-mode: earliest-offset
  properties:
- zookeeper.connect: $KAFKA_ZOOKEEPER_ADDRESS
  bootstrap.servers: $KAFKA_BOOTSTRAP_SERVERS
  format:
  type: json
@@ -86,8 +85,6 @@ tables:
  topic: test-avro
  startup-mode: earliest-offset
  properties:
- - key: zookeeper.connect
- value: $KAFKA_ZOOKEEPER_ADDRESS
  - key: bootstrap.servers
  value: $KAFKA_BOOTSTRAP_SERVERS
  format: