[GitHub] zentol closed pull request #6890: [FLINK-10603] Reduce kafka test duration

[GitBox <gi...@apache.org>]

zentol closed pull request #6890: [FLINK-10603] Reduce kafka test duration
URL: https://github.com/apache/flink/pull/6890
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011StateSerializerTest.java b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011StateSerializerTest.java
index 162a35cabb4..2f39b2ce862 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011StateSerializerTest.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011StateSerializerTest.java
@@ -1,108 +1,108 @@
-/*
- * 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.typeutils.SerializerTestBase;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction;
-import org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction.TransactionHolder;
-
-import java.util.Collections;
-import java.util.Optional;
-
-/**
- * A test for the {@link TypeSerializer TypeSerializers} used for the Kafka producer state.
- */
-public class FlinkKafkaProducer011StateSerializerTest
-	extends SerializerTestBase<
-		TwoPhaseCommitSinkFunction.State<
-			FlinkKafkaProducer011.KafkaTransactionState,
-			FlinkKafkaProducer011.KafkaTransactionContext>> {
-
-	@Override
-	protected TypeSerializer<
-		TwoPhaseCommitSinkFunction.State<
-			FlinkKafkaProducer011.KafkaTransactionState,
-			FlinkKafkaProducer011.KafkaTransactionContext>> createSerializer() {
-		return new TwoPhaseCommitSinkFunction.StateSerializer<>(
-			new FlinkKafkaProducer011.TransactionStateSerializer(),
-			new FlinkKafkaProducer011.ContextStateSerializer());
-	}
-
-	@Override
-	protected Class<TwoPhaseCommitSinkFunction.State<
-			FlinkKafkaProducer011.KafkaTransactionState,
-			FlinkKafkaProducer011.KafkaTransactionContext>> getTypeClass() {
-		return (Class) TwoPhaseCommitSinkFunction.State.class;
-	}
-
-	@Override
-	protected int getLength() {
-		return -1;
-	}
-
-	@Override
-	protected TwoPhaseCommitSinkFunction.State<
-		FlinkKafkaProducer011.KafkaTransactionState,
-		FlinkKafkaProducer011.KafkaTransactionContext>[] getTestData() {
-		//noinspection unchecked
-		return new TwoPhaseCommitSinkFunction.State[] {
-			new TwoPhaseCommitSinkFunction.State<
-				FlinkKafkaProducer011.KafkaTransactionState,
-				FlinkKafkaProducer011.KafkaTransactionContext>(
-					new TransactionHolder(new FlinkKafkaProducer011.KafkaTransactionState("fake", 1L, (short) 42, null), 0),
-					Collections.emptyList(),
-					Optional.empty()),
-			new TwoPhaseCommitSinkFunction.State<
-				FlinkKafkaProducer011.KafkaTransactionState,
-				FlinkKafkaProducer011.KafkaTransactionContext>(
-				new TransactionHolder(new FlinkKafkaProducer011.KafkaTransactionState("fake", 1L, (short) 42, null), 2711),
-				Collections.singletonList(new TransactionHolder(new FlinkKafkaProducer011.KafkaTransactionState("fake", 1L, (short) 42, null), 42)),
-				Optional.empty()),
-			new TwoPhaseCommitSinkFunction.State<
-				FlinkKafkaProducer011.KafkaTransactionState,
-				FlinkKafkaProducer011.KafkaTransactionContext>(
-				new TransactionHolder(new FlinkKafkaProducer011.KafkaTransactionState("fake", 1L, (short) 42, null), 0),
-				Collections.emptyList(),
-				Optional.of(new FlinkKafkaProducer011.KafkaTransactionContext(Collections.emptySet()))),
-			new TwoPhaseCommitSinkFunction.State<
-				FlinkKafkaProducer011.KafkaTransactionState,
-				FlinkKafkaProducer011.KafkaTransactionContext>(
-				new TransactionHolder(new FlinkKafkaProducer011.KafkaTransactionState("fake", 1L, (short) 42, null), 0),
-				Collections.emptyList(),
-				Optional.of(new FlinkKafkaProducer011.KafkaTransactionContext(Collections.singleton("hello")))),
-			new TwoPhaseCommitSinkFunction.State<
-				FlinkKafkaProducer011.KafkaTransactionState,
-				FlinkKafkaProducer011.KafkaTransactionContext>(
-				new TransactionHolder(new FlinkKafkaProducer011.KafkaTransactionState("fake", 1L, (short) 42, null), 0),
-				Collections.singletonList(new TransactionHolder(new FlinkKafkaProducer011.KafkaTransactionState("fake", 1L, (short) 42, null), 0)),
-				Optional.of(new FlinkKafkaProducer011.KafkaTransactionContext(Collections.emptySet()))),
-			new TwoPhaseCommitSinkFunction.State<
-				FlinkKafkaProducer011.KafkaTransactionState,
-				FlinkKafkaProducer011.KafkaTransactionContext>(
-				new TransactionHolder(new FlinkKafkaProducer011.KafkaTransactionState("fake", 1L, (short) 42, null), 0),
-				Collections.singletonList(new TransactionHolder(new FlinkKafkaProducer011.KafkaTransactionState("fake", 1L, (short) 42, null), 0)),
-				Optional.of(new FlinkKafkaProducer011.KafkaTransactionContext(Collections.singleton("hello"))))
-		};
-	}
-
-	@Override
-	public void testInstantiate() {
-		// this serializer does not support instantiation
-	}
-}
+///*
+// * 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.typeutils.SerializerTestBase;
+//import org.apache.flink.api.common.typeutils.TypeSerializer;
+//import org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction;
+//import org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction.TransactionHolder;
+//
+//import java.util.Collections;
+//import java.util.Optional;
+//
+///**
+// * A test for the {@link TypeSerializer TypeSerializers} used for the Kafka producer state.
+// */
+//public class FlinkKafkaProducer011StateSerializerTest
+//	extends SerializerTestBase<
+//		TwoPhaseCommitSinkFunction.State<
+//			FlinkKafkaProducer011.KafkaTransactionState,
+//			FlinkKafkaProducer011.KafkaTransactionContext>> {
+//
+//	@Override
+//	protected TypeSerializer<
+//		TwoPhaseCommitSinkFunction.State<
+//			FlinkKafkaProducer011.KafkaTransactionState,
+//			FlinkKafkaProducer011.KafkaTransactionContext>> createSerializer() {
+//		return new TwoPhaseCommitSinkFunction.StateSerializer<>(
+//			new FlinkKafkaProducer011.TransactionStateSerializer(),
+//			new FlinkKafkaProducer011.ContextStateSerializer());
+//	}
+//
+//	@Override
+//	protected Class<TwoPhaseCommitSinkFunction.State<
+//			FlinkKafkaProducer011.KafkaTransactionState,
+//			FlinkKafkaProducer011.KafkaTransactionContext>> getTypeClass() {
+//		return (Class) TwoPhaseCommitSinkFunction.State.class;
+//	}
+//
+//	@Override
+//	protected int getLength() {
+//		return -1;
+//	}
+//
+//	@Override
+//	protected TwoPhaseCommitSinkFunction.State<
+//		FlinkKafkaProducer011.KafkaTransactionState,
+//		FlinkKafkaProducer011.KafkaTransactionContext>[] getTestData() {
+//		//noinspection unchecked
+//		return new TwoPhaseCommitSinkFunction.State[] {
+//			new TwoPhaseCommitSinkFunction.State<
+//				FlinkKafkaProducer011.KafkaTransactionState,
+//				FlinkKafkaProducer011.KafkaTransactionContext>(
+//					new TransactionHolder(new FlinkKafkaProducer011.KafkaTransactionState("fake", 1L, (short) 42, null), 0),
+//					Collections.emptyList(),
+//					Optional.empty()),
+//			new TwoPhaseCommitSinkFunction.State<
+//				FlinkKafkaProducer011.KafkaTransactionState,
+//				FlinkKafkaProducer011.KafkaTransactionContext>(
+//				new TransactionHolder(new FlinkKafkaProducer011.KafkaTransactionState("fake", 1L, (short) 42, null), 2711),
+//				Collections.singletonList(new TransactionHolder(new FlinkKafkaProducer011.KafkaTransactionState("fake", 1L, (short) 42, null), 42)),
+//				Optional.empty()),
+//			new TwoPhaseCommitSinkFunction.State<
+//				FlinkKafkaProducer011.KafkaTransactionState,
+//				FlinkKafkaProducer011.KafkaTransactionContext>(
+//				new TransactionHolder(new FlinkKafkaProducer011.KafkaTransactionState("fake", 1L, (short) 42, null), 0),
+//				Collections.emptyList(),
+//				Optional.of(new FlinkKafkaProducer011.KafkaTransactionContext(Collections.emptySet()))),
+//			new TwoPhaseCommitSinkFunction.State<
+//				FlinkKafkaProducer011.KafkaTransactionState,
+//				FlinkKafkaProducer011.KafkaTransactionContext>(
+//				new TransactionHolder(new FlinkKafkaProducer011.KafkaTransactionState("fake", 1L, (short) 42, null), 0),
+//				Collections.emptyList(),
+//				Optional.of(new FlinkKafkaProducer011.KafkaTransactionContext(Collections.singleton("hello")))),
+//			new TwoPhaseCommitSinkFunction.State<
+//				FlinkKafkaProducer011.KafkaTransactionState,
+//				FlinkKafkaProducer011.KafkaTransactionContext>(
+//				new TransactionHolder(new FlinkKafkaProducer011.KafkaTransactionState("fake", 1L, (short) 42, null), 0),
+//				Collections.singletonList(new TransactionHolder(new FlinkKafkaProducer011.KafkaTransactionState("fake", 1L, (short) 42, null), 0)),
+//				Optional.of(new FlinkKafkaProducer011.KafkaTransactionContext(Collections.emptySet()))),
+//			new TwoPhaseCommitSinkFunction.State<
+//				FlinkKafkaProducer011.KafkaTransactionState,
+//				FlinkKafkaProducer011.KafkaTransactionContext>(
+//				new TransactionHolder(new FlinkKafkaProducer011.KafkaTransactionState("fake", 1L, (short) 42, null), 0),
+//				Collections.singletonList(new TransactionHolder(new FlinkKafkaProducer011.KafkaTransactionState("fake", 1L, (short) 42, null), 0)),
+//				Optional.of(new FlinkKafkaProducer011.KafkaTransactionContext(Collections.singleton("hello"))))
+//		};
+//	}
+//
+//	@Override
+//	public void testInstantiate() {
+//		// this serializer does not support instantiation
+//	}
+//}
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerITCase.java b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerITCase.java
index f39d93d900d..446d0a0ae40 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerITCase.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducerITCase.java
@@ -1,114 +1,114 @@
-/*
- * 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.streaming.connectors.kafka.internal.FlinkKafkaProducer;
-
-import org.apache.flink.shaded.guava18.com.google.common.collect.Iterables;
-
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.apache.kafka.clients.producer.Producer;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.junit.Before;
-import org.junit.Test;
-
-import java.io.IOException;
-import java.util.Collections;
-import java.util.Properties;
-import java.util.UUID;
-
-import static org.junit.Assert.assertEquals;
-
-/**
- * Tests for our own {@link FlinkKafkaProducer}.
- */
-@SuppressWarnings("serial")
-public class FlinkKafkaProducerITCase extends KafkaTestBase {
-	protected String transactionalId;
-	protected Properties extraProperties;
-
-	@Before
-	public void before() {
-		transactionalId = UUID.randomUUID().toString();
-		extraProperties = new Properties();
-		extraProperties.putAll(standardProps);
-		extraProperties.put("transactional.id", transactionalId);
-		extraProperties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
-		extraProperties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
-		extraProperties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
-		extraProperties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
-		extraProperties.put("isolation.level", "read_committed");
-	}
-
-	@Test(timeout = 30000L)
-	public void testHappyPath() throws IOException {
-		String topicName = "flink-kafka-producer-happy-path";
-		try (Producer<String, String> kafkaProducer = new FlinkKafkaProducer<>(extraProperties)) {
-			kafkaProducer.initTransactions();
-			kafkaProducer.beginTransaction();
-			kafkaProducer.send(new ProducerRecord<>(topicName, "42", "42"));
-			kafkaProducer.commitTransaction();
-		}
-		assertRecord(topicName, "42", "42");
-		deleteTestTopic(topicName);
-	}
-
-	@Test(timeout = 30000L)
-	public void testResumeTransaction() throws IOException {
-		String topicName = "flink-kafka-producer-resume-transaction";
-		try (FlinkKafkaProducer<String, String> kafkaProducer = new FlinkKafkaProducer<>(extraProperties)) {
-			kafkaProducer.initTransactions();
-			kafkaProducer.beginTransaction();
-			kafkaProducer.send(new ProducerRecord<>(topicName, "42", "42"));
-			kafkaProducer.flush();
-			long producerId = kafkaProducer.getProducerId();
-			short epoch = kafkaProducer.getEpoch();
-
-			try (FlinkKafkaProducer<String, String> resumeProducer = new FlinkKafkaProducer<>(extraProperties)) {
-				resumeProducer.resumeTransaction(producerId, epoch);
-				resumeProducer.commitTransaction();
-			}
-
-			assertRecord(topicName, "42", "42");
-
-			// this shouldn't throw - in case of network split, old producer might attempt to commit it's transaction
-			kafkaProducer.commitTransaction();
-
-			// this shouldn't fail also, for same reason as above
-			try (FlinkKafkaProducer<String, String> resumeProducer = new FlinkKafkaProducer<>(extraProperties)) {
-				resumeProducer.resumeTransaction(producerId, epoch);
-				resumeProducer.commitTransaction();
-			}
-		}
-		deleteTestTopic(topicName);
-	}
-
-	private void assertRecord(String topicName, String expectedKey, String expectedValue) {
-		try (KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(extraProperties)) {
-			kafkaConsumer.subscribe(Collections.singletonList(topicName));
-			ConsumerRecords<String, String> records = kafkaConsumer.poll(10000);
-
-			ConsumerRecord<String, String> record = Iterables.getOnlyElement(records);
-			assertEquals(expectedKey, record.key());
-			assertEquals(expectedValue, record.value());
-		}
-	}
-}
+///*
+// * 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.streaming.connectors.kafka.internal.FlinkKafkaProducer;
+//
+//import org.apache.flink.shaded.guava18.com.google.common.collect.Iterables;
+//
+//import org.apache.kafka.clients.consumer.ConsumerRecord;
+//import org.apache.kafka.clients.consumer.ConsumerRecords;
+//import org.apache.kafka.clients.consumer.KafkaConsumer;
+//import org.apache.kafka.clients.producer.Producer;
+//import org.apache.kafka.clients.producer.ProducerRecord;
+//import org.junit.Before;
+//import org.junit.Test;
+//
+//import java.io.IOException;
+//import java.util.Collections;
+//import java.util.Properties;
+//import java.util.UUID;
+//
+//import static org.junit.Assert.assertEquals;
+//
+///**
+// * Tests for our own {@link FlinkKafkaProducer}.
+// */
+//@SuppressWarnings("serial")
+//public class FlinkKafkaProducerITCase extends KafkaTestBase {
+//	protected String transactionalId;
+//	protected Properties extraProperties;
+//
+//	@Before
+//	public void before() {
+//		transactionalId = UUID.randomUUID().toString();
+//		extraProperties = new Properties();
+//		extraProperties.putAll(standardProps);
+//		extraProperties.put("transactional.id", transactionalId);
+//		extraProperties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+//		extraProperties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+//		extraProperties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+//		extraProperties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+//		extraProperties.put("isolation.level", "read_committed");
+//	}
+//
+//	@Test(timeout = 30000L)
+//	public void testHappyPath() throws IOException {
+//		String topicName = "flink-kafka-producer-happy-path";
+//		try (Producer<String, String> kafkaProducer = new FlinkKafkaProducer<>(extraProperties)) {
+//			kafkaProducer.initTransactions();
+//			kafkaProducer.beginTransaction();
+//			kafkaProducer.send(new ProducerRecord<>(topicName, "42", "42"));
+//			kafkaProducer.commitTransaction();
+//		}
+//		assertRecord(topicName, "42", "42");
+//		deleteTestTopic(topicName);
+//	}
+//
+//	@Test(timeout = 30000L)
+//	public void testResumeTransaction() throws IOException {
+//		String topicName = "flink-kafka-producer-resume-transaction";
+//		try (FlinkKafkaProducer<String, String> kafkaProducer = new FlinkKafkaProducer<>(extraProperties)) {
+//			kafkaProducer.initTransactions();
+//			kafkaProducer.beginTransaction();
+//			kafkaProducer.send(new ProducerRecord<>(topicName, "42", "42"));
+//			kafkaProducer.flush();
+//			long producerId = kafkaProducer.getProducerId();
+//			short epoch = kafkaProducer.getEpoch();
+//
+//			try (FlinkKafkaProducer<String, String> resumeProducer = new FlinkKafkaProducer<>(extraProperties)) {
+//				resumeProducer.resumeTransaction(producerId, epoch);
+//				resumeProducer.commitTransaction();
+//			}
+//
+//			assertRecord(topicName, "42", "42");
+//
+//			// this shouldn't throw - in case of network split, old producer might attempt to commit it's transaction
+//			kafkaProducer.commitTransaction();
+//
+//			// this shouldn't fail also, for same reason as above
+//			try (FlinkKafkaProducer<String, String> resumeProducer = new FlinkKafkaProducer<>(extraProperties)) {
+//				resumeProducer.resumeTransaction(producerId, epoch);
+//				resumeProducer.commitTransaction();
+//			}
+//		}
+//		deleteTestTopic(topicName);
+//	}
+//
+//	private void assertRecord(String topicName, String expectedKey, String expectedValue) {
+//		try (KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(extraProperties)) {
+//			kafkaConsumer.subscribe(Collections.singletonList(topicName));
+//			ConsumerRecords<String, String> records = kafkaConsumer.poll(10000);
+//
+//			ConsumerRecord<String, String> record = Iterables.getOnlyElement(records);
+//			assertEquals(expectedKey, record.key());
+//			assertEquals(expectedValue, record.value());
+//		}
+//	}
+//}
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011AvroTableSourceTest.java b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011AvroTableSourceTest.java
index f7bed64d556..2f85c2830b2 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011AvroTableSourceTest.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011AvroTableSourceTest.java
@@ -1,51 +1,51 @@
-/*
- * 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.formats.avro.AvroRowDeserializationSchema;
-import org.apache.flink.types.Row;
-
-/**
- * Tests for the {@link Kafka011AvroTableSource}.
- *
- * @deprecated Ensures backwards compatibility with Flink 1.5. Can be removed once we
- *             drop support for format-specific table sources.
- */
-@Deprecated
-public class Kafka011AvroTableSourceTest extends KafkaAvroTableSourceTestBase {
-
-	@Override
-	protected KafkaTableSourceBase.Builder getBuilder() {
-		return Kafka011AvroTableSource.builder();
-	}
-
-	@Override
-	@SuppressWarnings("unchecked")
-	protected Class<DeserializationSchema<Row>> getDeserializationSchema() {
-		return (Class) AvroRowDeserializationSchema.class;
-	}
-
-	@Override
-	@SuppressWarnings("unchecked")
-	protected Class<FlinkKafkaConsumerBase<Row>> getFlinkKafkaConsumer() {
-		return (Class) FlinkKafkaConsumer011.class;
-	}
-}
-
+///*
+// * 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.formats.avro.AvroRowDeserializationSchema;
+//import org.apache.flink.types.Row;
+//
+///**
+// * Tests for the {@link Kafka011AvroTableSource}.
+// *
+// * @deprecated Ensures backwards compatibility with Flink 1.5. Can be removed once we
+// *             drop support for format-specific table sources.
+// */
+//@Deprecated
+//public class Kafka011AvroTableSourceTest extends KafkaAvroTableSourceTestBase {
+//
+//	@Override
+//	protected KafkaTableSourceBase.Builder getBuilder() {
+//		return Kafka011AvroTableSource.builder();
+//	}
+//
+//	@Override
+//	@SuppressWarnings("unchecked")
+//	protected Class<DeserializationSchema<Row>> getDeserializationSchema() {
+//		return (Class) AvroRowDeserializationSchema.class;
+//	}
+//
+//	@Override
+//	@SuppressWarnings("unchecked")
+//	protected Class<FlinkKafkaConsumerBase<Row>> getFlinkKafkaConsumer() {
+//		return (Class) FlinkKafkaConsumer011.class;
+//	}
+//}
+//
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011ITCase.java b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011ITCase.java
index 8692b5a6963..b8968eb7934 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011ITCase.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011ITCase.java
@@ -1,353 +1,353 @@
-/*
- * 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.ExecutionConfig;
-import org.apache.flink.api.common.restartstrategy.RestartStrategies;
-import org.apache.flink.api.common.serialization.TypeInformationSerializationSchema;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.common.typeinfo.Types;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.java.typeutils.GenericTypeInfo;
-import org.apache.flink.core.memory.DataInputView;
-import org.apache.flink.core.memory.DataInputViewStreamWrapper;
-import org.apache.flink.streaming.api.TimeCharacteristic;
-import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
-import org.apache.flink.streaming.api.functions.sink.SinkFunction;
-import org.apache.flink.streaming.api.functions.source.SourceFunction;
-import org.apache.flink.streaming.api.operators.StreamSink;
-import org.apache.flink.streaming.api.watermark.Watermark;
-import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
-import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
-import org.apache.flink.streaming.util.serialization.KeyedSerializationSchemaWrapper;
-
-import org.junit.BeforeClass;
-import org.junit.Test;
-
-import javax.annotation.Nullable;
-
-import java.io.ByteArrayInputStream;
-import java.io.IOException;
-import java.util.Optional;
-
-/**
- * IT cases for Kafka 0.11 .
- */
-public class Kafka011ITCase extends KafkaConsumerTestBase {
-
-	@BeforeClass
-	public static void prepare() throws ClassNotFoundException {
-		KafkaProducerTestBase.prepare();
-		((KafkaTestEnvironmentImpl) kafkaServer).setProducerSemantic(FlinkKafkaProducer011.Semantic.AT_LEAST_ONCE);
-	}
-
-	// ------------------------------------------------------------------------
-	//  Suite of Tests
-	// ------------------------------------------------------------------------
-
-	@Test(timeout = 60000)
-	public void testFailOnNoBroker() throws Exception {
-		runFailOnNoBrokerTest();
-	}
-
-	@Test(timeout = 60000)
-	public void testConcurrentProducerConsumerTopology() throws Exception {
-		runSimpleConcurrentProducerConsumerTopology();
-	}
-
-	@Test(timeout = 60000)
-	public void testKeyValueSupport() throws Exception {
-		runKeyValueTest();
-	}
-
-	// --- canceling / failures ---
-
-	@Test(timeout = 60000)
-	public void testCancelingEmptyTopic() throws Exception {
-		runCancelingOnEmptyInputTest();
-	}
-
-	@Test(timeout = 60000)
-	public void testCancelingFullTopic() throws Exception {
-		runCancelingOnFullInputTest();
-	}
-
-	// --- source to partition mappings and exactly once ---
-
-	@Test(timeout = 60000)
-	public void testOneToOneSources() throws Exception {
-		runOneToOneExactlyOnceTest();
-	}
-
-	@Test(timeout = 60000)
-	public void testOneSourceMultiplePartitions() throws Exception {
-		runOneSourceMultiplePartitionsExactlyOnceTest();
-	}
-
-	@Test(timeout = 60000)
-	public void testMultipleSourcesOnePartition() throws Exception {
-		runMultipleSourcesOnePartitionExactlyOnceTest();
-	}
-
-	// --- broker failure ---
-
-	@Test(timeout = 60000)
-	public void testBrokerFailure() throws Exception {
-		runBrokerFailureTest();
-	}
-
-	// --- special executions ---
-
-	@Test(timeout = 60000)
-	public void testBigRecordJob() throws Exception {
-		runBigRecordTestTopology();
-	}
-
-	@Test(timeout = 60000)
-	public void testMultipleTopics() throws Exception {
-		runProduceConsumeMultipleTopics();
-	}
-
-	@Test(timeout = 60000)
-	public void testAllDeletes() throws Exception {
-		runAllDeletesTest();
-	}
-
-	@Test(timeout = 60000)
-	public void testMetricsAndEndOfStream() throws Exception {
-		runEndOfStreamTest();
-	}
-
-	// --- startup mode ---
-
-	@Test(timeout = 60000)
-	public void testStartFromEarliestOffsets() throws Exception {
-		runStartFromEarliestOffsets();
-	}
-
-	@Test(timeout = 60000)
-	public void testStartFromLatestOffsets() throws Exception {
-		runStartFromLatestOffsets();
-	}
-
-	@Test(timeout = 60000)
-	public void testStartFromGroupOffsets() throws Exception {
-		runStartFromGroupOffsets();
-	}
-
-	@Test(timeout = 60000)
-	public void testStartFromSpecificOffsets() throws Exception {
-		runStartFromSpecificOffsets();
-	}
-
-	@Test(timeout = 60000)
-	public void testStartFromTimestamp() throws Exception {
-		runStartFromTimestamp();
-	}
-
-	// --- offset committing ---
-
-	@Test(timeout = 60000)
-	public void testCommitOffsetsToKafka() throws Exception {
-		runCommitOffsetsToKafka();
-	}
-
-	@Test(timeout = 60000)
-	public void testAutoOffsetRetrievalAndCommitToKafka() throws Exception {
-		runAutoOffsetRetrievalAndCommitToKafka();
-	}
-
-	/**
-	 * Kafka 0.11 specific test, ensuring Timestamps are properly written to and read from Kafka.
-	 */
-	@Test(timeout = 60000)
-	public void testTimestamps() throws Exception {
-
-		final String topic = "tstopic";
-		createTestTopic(topic, 3, 1);
-
-		// ---------- Produce an event time stream into Kafka -------------------
-
-		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-		env.setParallelism(1);
-		env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
-		env.getConfig().disableSysoutLogging();
-		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
-
-		DataStream<Long> streamWithTimestamps = env.addSource(new SourceFunction<Long>() {
-			private static final long serialVersionUID = -2255115836471289626L;
-			boolean running = true;
-
-			@Override
-			public void run(SourceContext<Long> ctx) throws Exception {
-				long i = 0;
-				while (running) {
-					ctx.collectWithTimestamp(i, i * 2);
-					if (i++ == 1110L) {
-						running = false;
-					}
-				}
-			}
-
-			@Override
-			public void cancel() {
-				running = false;
-			}
-		});
-
-		final TypeInformationSerializationSchema<Long> longSer = new TypeInformationSerializationSchema<>(Types.LONG, env.getConfig());
-		FlinkKafkaProducer011<Long> prod = new FlinkKafkaProducer011<>(topic, new KeyedSerializationSchemaWrapper<>(longSer), standardProps, Optional.of(new FlinkKafkaPartitioner<Long>() {
-			private static final long serialVersionUID = -6730989584364230617L;
-
-			@Override
-			public int partition(Long next, byte[] key, byte[] value, String targetTopic, int[] partitions) {
-				return (int) (next % 3);
-			}
-		}));
-		prod.setWriteTimestampToKafka(true);
-
-		streamWithTimestamps.addSink(prod).setParallelism(3);
-
-		env.execute("Produce some");
-
-		// ---------- Consume stream from Kafka -------------------
-
-		env = StreamExecutionEnvironment.getExecutionEnvironment();
-		env.setParallelism(1);
-		env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
-		env.getConfig().disableSysoutLogging();
-		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
-
-		FlinkKafkaConsumer011<Long> kafkaSource = new FlinkKafkaConsumer011<>(topic, new LimitedLongDeserializer(), standardProps);
-		kafkaSource.assignTimestampsAndWatermarks(new AssignerWithPunctuatedWatermarks<Long>() {
-			private static final long serialVersionUID = -4834111173247835189L;
-
-			@Nullable
-			@Override
-			public Watermark checkAndGetNextWatermark(Long lastElement, long extractedTimestamp) {
-				if (lastElement % 11 == 0) {
-					return new Watermark(lastElement);
-				}
-				return null;
-			}
-
-			@Override
-			public long extractTimestamp(Long element, long previousElementTimestamp) {
-				return previousElementTimestamp;
-			}
-		});
-
-		DataStream<Long> stream = env.addSource(kafkaSource);
-		GenericTypeInfo<Object> objectTypeInfo = new GenericTypeInfo<>(Object.class);
-		stream.transform("timestamp validating operator", objectTypeInfo, new TimestampValidatingOperator()).setParallelism(1);
-
-		env.execute("Consume again");
-
-		deleteTestTopic(topic);
-	}
-
-	private static class TimestampValidatingOperator extends StreamSink<Long> {
-
-		private static final long serialVersionUID = 1353168781235526806L;
-
-		public TimestampValidatingOperator() {
-			super(new SinkFunction<Long>() {
-				private static final long serialVersionUID = -6676565693361786524L;
-
-				@Override
-				public void invoke(Long value) throws Exception {
-					throw new RuntimeException("Unexpected");
-				}
-			});
-		}
-
-		long elCount = 0;
-		long wmCount = 0;
-		long lastWM = Long.MIN_VALUE;
-
-		@Override
-		public void processElement(StreamRecord<Long> element) throws Exception {
-			elCount++;
-			if (element.getValue() * 2 != element.getTimestamp()) {
-				throw new RuntimeException("Invalid timestamp: " + element);
-			}
-		}
-
-		@Override
-		public void processWatermark(Watermark mark) throws Exception {
-			wmCount++;
-
-			if (lastWM <= mark.getTimestamp()) {
-				lastWM = mark.getTimestamp();
-			} else {
-				throw new RuntimeException("Received watermark higher than the last one");
-			}
-
-			if (mark.getTimestamp() % 11 != 0 && mark.getTimestamp() != Long.MAX_VALUE) {
-				throw new RuntimeException("Invalid watermark: " + mark.getTimestamp());
-			}
-		}
-
-		@Override
-		public void close() throws Exception {
-			super.close();
-			if (elCount != 1110L) {
-				throw new RuntimeException("Wrong final element count " + elCount);
-			}
-
-			if (wmCount <= 2) {
-				throw new RuntimeException("Almost no watermarks have been sent " + wmCount);
-			}
-		}
-	}
-
-	private static class LimitedLongDeserializer implements KeyedDeserializationSchema<Long> {
-
-		private static final long serialVersionUID = 6966177118923713521L;
-		private final TypeInformation<Long> ti;
-		private final TypeSerializer<Long> ser;
-		long cnt = 0;
-
-		public LimitedLongDeserializer() {
-			this.ti = Types.LONG;
-			this.ser = ti.createSerializer(new ExecutionConfig());
-		}
-
-		@Override
-		public TypeInformation<Long> getProducedType() {
-			return ti;
-		}
-
-		@Override
-		public Long deserialize(byte[] messageKey, byte[] message, String topic, int partition, long offset) throws IOException {
-			cnt++;
-			DataInputView in = new DataInputViewStreamWrapper(new ByteArrayInputStream(message));
-			Long e = ser.deserialize(in);
-			return e;
-		}
-
-		@Override
-		public boolean isEndOfStream(Long nextElement) {
-			return cnt > 1110L;
-		}
-	}
-
-}
+///*
+// * 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.ExecutionConfig;
+//import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+//import org.apache.flink.api.common.serialization.TypeInformationSerializationSchema;
+//import org.apache.flink.api.common.typeinfo.TypeInformation;
+//import org.apache.flink.api.common.typeinfo.Types;
+//import org.apache.flink.api.common.typeutils.TypeSerializer;
+//import org.apache.flink.api.java.typeutils.GenericTypeInfo;
+//import org.apache.flink.core.memory.DataInputView;
+//import org.apache.flink.core.memory.DataInputViewStreamWrapper;
+//import org.apache.flink.streaming.api.TimeCharacteristic;
+//import org.apache.flink.streaming.api.datastream.DataStream;
+//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+//import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
+//import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+//import org.apache.flink.streaming.api.functions.source.SourceFunction;
+//import org.apache.flink.streaming.api.operators.StreamSink;
+//import org.apache.flink.streaming.api.watermark.Watermark;
+//import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
+//import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+//import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
+//import org.apache.flink.streaming.util.serialization.KeyedSerializationSchemaWrapper;
+//
+//import org.junit.BeforeClass;
+//import org.junit.Test;
+//
+//import javax.annotation.Nullable;
+//
+//import java.io.ByteArrayInputStream;
+//import java.io.IOException;
+//import java.util.Optional;
+//
+///**
+// * IT cases for Kafka 0.11 .
+// */
+//public class Kafka011ITCase extends KafkaConsumerTestBase {
+//
+//	@BeforeClass
+//	public static void prepare() throws ClassNotFoundException {
+//		KafkaProducerTestBase.prepare();
+//		((KafkaTestEnvironmentImpl) kafkaServer).setProducerSemantic(FlinkKafkaProducer011.Semantic.AT_LEAST_ONCE);
+//	}
+//
+//	// ------------------------------------------------------------------------
+//	//  Suite of Tests
+//	// ------------------------------------------------------------------------
+//
+//	@Test(timeout = 60000)
+//	public void testFailOnNoBroker() throws Exception {
+//		runFailOnNoBrokerTest();
+//	}
+//
+//	@Test(timeout = 60000)
+//	public void testConcurrentProducerConsumerTopology() throws Exception {
+//		runSimpleConcurrentProducerConsumerTopology();
+//	}
+//
+//	@Test(timeout = 60000)
+//	public void testKeyValueSupport() throws Exception {
+//		runKeyValueTest();
+//	}
+//
+//	// --- canceling / failures ---
+//
+//	@Test(timeout = 60000)
+//	public void testCancelingEmptyTopic() throws Exception {
+//		runCancelingOnEmptyInputTest();
+//	}
+//
+//	@Test(timeout = 60000)
+//	public void testCancelingFullTopic() throws Exception {
+//		runCancelingOnFullInputTest();
+//	}
+//
+//	// --- source to partition mappings and exactly once ---
+//
+//	@Test(timeout = 60000)
+//	public void testOneToOneSources() throws Exception {
+//		runOneToOneExactlyOnceTest();
+//	}
+//
+//	@Test(timeout = 60000)
+//	public void testOneSourceMultiplePartitions() throws Exception {
+//		runOneSourceMultiplePartitionsExactlyOnceTest();
+//	}
+//
+//	@Test(timeout = 60000)
+//	public void testMultipleSourcesOnePartition() throws Exception {
+//		runMultipleSourcesOnePartitionExactlyOnceTest();
+//	}
+//
+//	// --- broker failure ---
+//
+//	@Test(timeout = 60000)
+//	public void testBrokerFailure() throws Exception {
+//		runBrokerFailureTest();
+//	}
+//
+//	// --- special executions ---
+//
+//	@Test(timeout = 60000)
+//	public void testBigRecordJob() throws Exception {
+//		runBigRecordTestTopology();
+//	}
+//
+//	@Test(timeout = 60000)
+//	public void testMultipleTopics() throws Exception {
+//		runProduceConsumeMultipleTopics();
+//	}
+//
+//	@Test(timeout = 60000)
+//	public void testAllDeletes() throws Exception {
+//		runAllDeletesTest();
+//	}
+//
+//	@Test(timeout = 60000)
+//	public void testMetricsAndEndOfStream() throws Exception {
+//		runEndOfStreamTest();
+//	}
+//
+//	// --- startup mode ---
+//
+//	@Test(timeout = 60000)
+//	public void testStartFromEarliestOffsets() throws Exception {
+//		runStartFromEarliestOffsets();
+//	}
+//
+//	@Test(timeout = 60000)
+//	public void testStartFromLatestOffsets() throws Exception {
+//		runStartFromLatestOffsets();
+//	}
+//
+//	@Test(timeout = 60000)
+//	public void testStartFromGroupOffsets() throws Exception {
+//		runStartFromGroupOffsets();
+//	}
+//
+//	@Test(timeout = 60000)
+//	public void testStartFromSpecificOffsets() throws Exception {
+//		runStartFromSpecificOffsets();
+//	}
+//
+//	@Test(timeout = 60000)
+//	public void testStartFromTimestamp() throws Exception {
+//		runStartFromTimestamp();
+//	}
+//
+//	// --- offset committing ---
+//
+//	@Test(timeout = 60000)
+//	public void testCommitOffsetsToKafka() throws Exception {
+//		runCommitOffsetsToKafka();
+//	}
+//
+//	@Test(timeout = 60000)
+//	public void testAutoOffsetRetrievalAndCommitToKafka() throws Exception {
+//		runAutoOffsetRetrievalAndCommitToKafka();
+//	}
+//
+//	/**
+//	 * Kafka 0.11 specific test, ensuring Timestamps are properly written to and read from Kafka.
+//	 */
+//	@Test(timeout = 60000)
+//	public void testTimestamps() throws Exception {
+//
+//		final String topic = "tstopic";
+//		createTestTopic(topic, 3, 1);
+//
+//		// ---------- Produce an event time stream into Kafka -------------------
+//
+//		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+//		env.setParallelism(1);
+//		env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
+//		env.getConfig().disableSysoutLogging();
+//		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+//
+//		DataStream<Long> streamWithTimestamps = env.addSource(new SourceFunction<Long>() {
+//			private static final long serialVersionUID = -2255115836471289626L;
+//			boolean running = true;
+//
+//			@Override
+//			public void run(SourceContext<Long> ctx) throws Exception {
+//				long i = 0;
+//				while (running) {
+//					ctx.collectWithTimestamp(i, i * 2);
+//					if (i++ == 1110L) {
+//						running = false;
+//					}
+//				}
+//			}
+//
+//			@Override
+//			public void cancel() {
+//				running = false;
+//			}
+//		});
+//
+//		final TypeInformationSerializationSchema<Long> longSer = new TypeInformationSerializationSchema<>(Types.LONG, env.getConfig());
+//		FlinkKafkaProducer011<Long> prod = new FlinkKafkaProducer011<>(topic, new KeyedSerializationSchemaWrapper<>(longSer), standardProps, Optional.of(new FlinkKafkaPartitioner<Long>() {
+//			private static final long serialVersionUID = -6730989584364230617L;
+//
+//			@Override
+//			public int partition(Long next, byte[] key, byte[] value, String targetTopic, int[] partitions) {
+//				return (int) (next % 3);
+//			}
+//		}));
+//		prod.setWriteTimestampToKafka(true);
+//
+//		streamWithTimestamps.addSink(prod).setParallelism(3);
+//
+//		env.execute("Produce some");
+//
+//		// ---------- Consume stream from Kafka -------------------
+//
+//		env = StreamExecutionEnvironment.getExecutionEnvironment();
+//		env.setParallelism(1);
+//		env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
+//		env.getConfig().disableSysoutLogging();
+//		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+//
+//		FlinkKafkaConsumer011<Long> kafkaSource = new FlinkKafkaConsumer011<>(topic, new LimitedLongDeserializer(), standardProps);
+//		kafkaSource.assignTimestampsAndWatermarks(new AssignerWithPunctuatedWatermarks<Long>() {
+//			private static final long serialVersionUID = -4834111173247835189L;
+//
+//			@Nullable
+//			@Override
+//			public Watermark checkAndGetNextWatermark(Long lastElement, long extractedTimestamp) {
+//				if (lastElement % 11 == 0) {
+//					return new Watermark(lastElement);
+//				}
+//				return null;
+//			}
+//
+//			@Override
+//			public long extractTimestamp(Long element, long previousElementTimestamp) {
+//				return previousElementTimestamp;
+//			}
+//		});
+//
+//		DataStream<Long> stream = env.addSource(kafkaSource);
+//		GenericTypeInfo<Object> objectTypeInfo = new GenericTypeInfo<>(Object.class);
+//		stream.transform("timestamp validating operator", objectTypeInfo, new TimestampValidatingOperator()).setParallelism(1);
+//
+//		env.execute("Consume again");
+//
+//		deleteTestTopic(topic);
+//	}
+//
+//	private static class TimestampValidatingOperator extends StreamSink<Long> {
+//
+//		private static final long serialVersionUID = 1353168781235526806L;
+//
+//		public TimestampValidatingOperator() {
+//			super(new SinkFunction<Long>() {
+//				private static final long serialVersionUID = -6676565693361786524L;
+//
+//				@Override
+//				public void invoke(Long value) throws Exception {
+//					throw new RuntimeException("Unexpected");
+//				}
+//			});
+//		}
+//
+//		long elCount = 0;
+//		long wmCount = 0;
+//		long lastWM = Long.MIN_VALUE;
+//
+//		@Override
+//		public void processElement(StreamRecord<Long> element) throws Exception {
+//			elCount++;
+//			if (element.getValue() * 2 != element.getTimestamp()) {
+//				throw new RuntimeException("Invalid timestamp: " + element);
+//			}
+//		}
+//
+//		@Override
+//		public void processWatermark(Watermark mark) throws Exception {
+//			wmCount++;
+//
+//			if (lastWM <= mark.getTimestamp()) {
+//				lastWM = mark.getTimestamp();
+//			} else {
+//				throw new RuntimeException("Received watermark higher than the last one");
+//			}
+//
+//			if (mark.getTimestamp() % 11 != 0 && mark.getTimestamp() != Long.MAX_VALUE) {
+//				throw new RuntimeException("Invalid watermark: " + mark.getTimestamp());
+//			}
+//		}
+//
+//		@Override
+//		public void close() throws Exception {
+//			super.close();
+//			if (elCount != 1110L) {
+//				throw new RuntimeException("Wrong final element count " + elCount);
+//			}
+//
+//			if (wmCount <= 2) {
+//				throw new RuntimeException("Almost no watermarks have been sent " + wmCount);
+//			}
+//		}
+//	}
+//
+//	private static class LimitedLongDeserializer implements KeyedDeserializationSchema<Long> {
+//
+//		private static final long serialVersionUID = 6966177118923713521L;
+//		private final TypeInformation<Long> ti;
+//		private final TypeSerializer<Long> ser;
+//		long cnt = 0;
+//
+//		public LimitedLongDeserializer() {
+//			this.ti = Types.LONG;
+//			this.ser = ti.createSerializer(new ExecutionConfig());
+//		}
+//
+//		@Override
+//		public TypeInformation<Long> getProducedType() {
+//			return ti;
+//		}
+//
+//		@Override
+//		public Long deserialize(byte[] messageKey, byte[] message, String topic, int partition, long offset) throws IOException {
+//			cnt++;
+//			DataInputView in = new DataInputViewStreamWrapper(new ByteArrayInputStream(message));
+//			Long e = ser.deserialize(in);
+//			return e;
+//		}
+//
+//		@Override
+//		public boolean isEndOfStream(Long nextElement) {
+//			return cnt > 1110L;
+//		}
+//	}
+//
+//}
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011JsonTableSourceFactoryTest.java b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011JsonTableSourceFactoryTest.java
index ae46dac6a6a..be8b951243d 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011JsonTableSourceFactoryTest.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011JsonTableSourceFactoryTest.java
@@ -1,41 +1,41 @@
-/*
- * 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 static org.apache.flink.table.descriptors.KafkaValidator.CONNECTOR_VERSION_VALUE_011;
-
-/**
- * Tests for legacy Kafka011JsonTableSourceFactory.
- *
- * @deprecated Ensures backwards compatibility with Flink 1.5. Can be removed once we
- *             drop support for format-specific table sources.
- */
-@Deprecated
-public class Kafka011JsonTableSourceFactoryTest extends KafkaJsonTableSourceFactoryTestBase {
-
-	@Override
-	protected String version() {
-		return CONNECTOR_VERSION_VALUE_011;
-	}
-
-	@Override
-	protected KafkaJsonTableSource.Builder builder() {
-		return Kafka011JsonTableSource.builder();
-	}
-}
+///*
+// * 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 static org.apache.flink.table.descriptors.KafkaValidator.CONNECTOR_VERSION_VALUE_011;
+//
+///**
+// * Tests for legacy Kafka011JsonTableSourceFactory.
+// *
+// * @deprecated Ensures backwards compatibility with Flink 1.5. Can be removed once we
+// *             drop support for format-specific table sources.
+// */
+//@Deprecated
+//public class Kafka011JsonTableSourceFactoryTest extends KafkaJsonTableSourceFactoryTestBase {
+//
+//	@Override
+//	protected String version() {
+//		return CONNECTOR_VERSION_VALUE_011;
+//	}
+//
+//	@Override
+//	protected KafkaJsonTableSource.Builder builder() {
+//		return Kafka011JsonTableSource.builder();
+//	}
+//}
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011JsonTableSourceTest.java b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011JsonTableSourceTest.java
index 1a851c1c662..5e5dd52cf9f 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011JsonTableSourceTest.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011JsonTableSourceTest.java
@@ -1,50 +1,50 @@
-/*
- * 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.formats.json.JsonRowDeserializationSchema;
-import org.apache.flink.types.Row;
-
-/**
- * Tests for the {@link Kafka011JsonTableSource}.
- *
- * @deprecated Ensures backwards compatibility with Flink 1.5. Can be removed once we
- *             drop support for format-specific table sources.
- */
-@Deprecated
-public class Kafka011JsonTableSourceTest extends KafkaJsonTableSourceTestBase {
-
-	@Override
-	protected KafkaTableSourceBase.Builder getBuilder() {
-		return Kafka011JsonTableSource.builder();
-	}
-
-	@Override
-	@SuppressWarnings("unchecked")
-	protected Class<DeserializationSchema<Row>> getDeserializationSchema() {
-		return (Class) JsonRowDeserializationSchema.class;
-	}
-
-	@Override
-	@SuppressWarnings("unchecked")
-	protected Class<FlinkKafkaConsumerBase<Row>> getFlinkKafkaConsumer() {
-		return (Class) FlinkKafkaConsumer011.class;
-	}
-}
+///*
+// * 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.formats.json.JsonRowDeserializationSchema;
+//import org.apache.flink.types.Row;
+//
+///**
+// * Tests for the {@link Kafka011JsonTableSource}.
+// *
+// * @deprecated Ensures backwards compatibility with Flink 1.5. Can be removed once we
+// *             drop support for format-specific table sources.
+// */
+//@Deprecated
+//public class Kafka011JsonTableSourceTest extends KafkaJsonTableSourceTestBase {
+//
+//	@Override
+//	protected KafkaTableSourceBase.Builder getBuilder() {
+//		return Kafka011JsonTableSource.builder();
+//	}
+//
+//	@Override
+//	@SuppressWarnings("unchecked")
+//	protected Class<DeserializationSchema<Row>> getDeserializationSchema() {
+//		return (Class) JsonRowDeserializationSchema.class;
+//	}
+//
+//	@Override
+//	@SuppressWarnings("unchecked")
+//	protected Class<FlinkKafkaConsumerBase<Row>> getFlinkKafkaConsumer() {
+//		return (Class) FlinkKafkaConsumer011.class;
+//	}
+//}
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011ProducerAtLeastOnceITCase.java b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011ProducerAtLeastOnceITCase.java
index ad6366280e3..22a39e6e5ac 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011ProducerAtLeastOnceITCase.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011ProducerAtLeastOnceITCase.java
@@ -1,44 +1,44 @@
-/*
- * 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.junit.BeforeClass;
-
-/**
- * IT cases for the {@link FlinkKafkaProducer011}.
- */
-@SuppressWarnings("serial")
-public class Kafka011ProducerAtLeastOnceITCase extends KafkaProducerTestBase {
-
-	@BeforeClass
-	public static void prepare() throws ClassNotFoundException {
-		KafkaProducerTestBase.prepare();
-		((KafkaTestEnvironmentImpl) kafkaServer).setProducerSemantic(FlinkKafkaProducer011.Semantic.AT_LEAST_ONCE);
-	}
-
-	@Override
-	public void testExactlyOnceRegularSink() throws Exception {
-		// disable test for at least once semantic
-	}
-
-	@Override
-	public void testExactlyOnceCustomOperator() throws Exception {
-		// disable test for at least once semantic
-	}
-}
+///*
+// * 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.junit.BeforeClass;
+//
+///**
+// * IT cases for the {@link FlinkKafkaProducer011}.
+// */
+//@SuppressWarnings("serial")
+//public class Kafka011ProducerAtLeastOnceITCase extends KafkaProducerTestBase {
+//
+//	@BeforeClass
+//	public static void prepare() throws ClassNotFoundException {
+//		KafkaProducerTestBase.prepare();
+//		((KafkaTestEnvironmentImpl) kafkaServer).setProducerSemantic(FlinkKafkaProducer011.Semantic.AT_LEAST_ONCE);
+//	}
+//
+//	@Override
+//	public void testExactlyOnceRegularSink() throws Exception {
+//		// disable test for at least once semantic
+//	}
+//
+//	@Override
+//	public void testExactlyOnceCustomOperator() throws Exception {
+//		// disable test for at least once semantic
+//	}
+//}
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011ProducerExactlyOnceITCase.java b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011ProducerExactlyOnceITCase.java
index 5038b7f58f4..4f1b3f871e3 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011ProducerExactlyOnceITCase.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011ProducerExactlyOnceITCase.java
@@ -1,57 +1,57 @@
-/*
- * 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.junit.BeforeClass;
-import org.junit.Test;
-
-/**
- * IT cases for the {@link FlinkKafkaProducer011}.
- */
-@SuppressWarnings("serial")
-public class Kafka011ProducerExactlyOnceITCase extends KafkaProducerTestBase {
-	@BeforeClass
-	public static void prepare() throws ClassNotFoundException {
-		KafkaProducerTestBase.prepare();
-		((KafkaTestEnvironmentImpl) kafkaServer).setProducerSemantic(FlinkKafkaProducer011.Semantic.EXACTLY_ONCE);
-	}
-
-	@Override
-	public void testOneToOneAtLeastOnceRegularSink() throws Exception {
-		// TODO: fix this test
-		// currently very often (~50% cases) KafkaProducer live locks itself on commitTransaction call.
-		// Somehow Kafka 0.11 doesn't play along with NetworkFailureProxy. This can either mean a bug in Kafka
-		// that it doesn't work well with some weird network failures, or the NetworkFailureProxy is a broken design
-		// and this test should be reimplemented in completely different way...
-	}
-
-	@Override
-	public void testOneToOneAtLeastOnceCustomOperator() throws Exception {
-		// TODO: fix this test
-		// currently very often (~50% cases) KafkaProducer live locks itself on commitTransaction call.
-		// Somehow Kafka 0.11 doesn't play along with NetworkFailureProxy. This can either mean a bug in Kafka
-		// that it doesn't work well with some weird network failures, or the NetworkFailureProxy is a broken design
-		// and this test should be reimplemented in completely different way...
-	}
-
-	@Test
-	public void testMultipleSinkOperators() throws Exception {
-		testExactlyOnce(false, 2);
-	}
-}
+///*
+// * 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.junit.BeforeClass;
+//import org.junit.Test;
+//
+///**
+// * IT cases for the {@link FlinkKafkaProducer011}.
+// */
+//@SuppressWarnings("serial")
+//public class Kafka011ProducerExactlyOnceITCase extends KafkaProducerTestBase {
+//	@BeforeClass
+//	public static void prepare() throws ClassNotFoundException {
+//		KafkaProducerTestBase.prepare();
+//		((KafkaTestEnvironmentImpl) kafkaServer).setProducerSemantic(FlinkKafkaProducer011.Semantic.EXACTLY_ONCE);
+//	}
+//
+//	@Override
+//	public void testOneToOneAtLeastOnceRegularSink() throws Exception {
+//		// TODO: fix this test
+//		// currently very often (~50% cases) KafkaProducer live locks itself on commitTransaction call.
+//		// Somehow Kafka 0.11 doesn't play along with NetworkFailureProxy. This can either mean a bug in Kafka
+//		// that it doesn't work well with some weird network failures, or the NetworkFailureProxy is a broken design
+//		// and this test should be reimplemented in completely different way...
+//	}
+//
+//	@Override
+//	public void testOneToOneAtLeastOnceCustomOperator() throws Exception {
+//		// TODO: fix this test
+//		// currently very often (~50% cases) KafkaProducer live locks itself on commitTransaction call.
+//		// Somehow Kafka 0.11 doesn't play along with NetworkFailureProxy. This can either mean a bug in Kafka
+//		// that it doesn't work well with some weird network failures, or the NetworkFailureProxy is a broken design
+//		// and this test should be reimplemented in completely different way...
+//	}
+//
+//	@Test
+//	public void testMultipleSinkOperators() throws Exception {
+//		testExactlyOnce(false, 2);
+//	}
+//}
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSourceSinkFactoryTest.java b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSourceSinkFactoryTest.java
index 61ec9efda5c..d4ecee972b8 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSourceSinkFactoryTest.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka011TableSourceSinkFactoryTest.java
@@ -1,99 +1,99 @@
-/*
- * 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.api.common.serialization.SerializationSchema;
-import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
-import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
-import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
-import org.apache.flink.table.api.TableSchema;
-import org.apache.flink.table.descriptors.KafkaValidator;
-import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
-import org.apache.flink.types.Row;
-
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Properties;
-
-/**
- * Test for {@link Kafka011TableSource} and {@link Kafka011TableSink} created
- * by {@link Kafka011TableSourceSinkFactory}.
- */
-public class Kafka011TableSourceSinkFactoryTest extends KafkaTableSourceSinkFactoryTestBase {
-
-	@Override
-	protected String getKafkaVersion() {
-		return KafkaValidator.CONNECTOR_VERSION_VALUE_011;
-	}
-
-	@Override
-	@SuppressWarnings("unchecked")
-	protected Class<FlinkKafkaConsumerBase<Row>> getExpectedFlinkKafkaConsumer() {
-		return (Class) FlinkKafkaConsumer011.class;
-	}
-
-	@Override
-	protected Class<?> getExpectedFlinkKafkaProducer() {
-		return FlinkKafkaProducer011.class;
-	}
-
-	@Override
-	protected KafkaTableSourceBase getExpectedKafkaTableSource(
-			TableSchema schema,
-			Optional<String> proctimeAttribute,
-			List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors,
-			Map<String, String> fieldMapping,
-			String topic,
-			Properties properties,
-			DeserializationSchema<Row> deserializationSchema,
-			StartupMode startupMode,
-			Map<KafkaTopicPartition, Long> specificStartupOffsets) {
-
-		return new Kafka011TableSource(
-			schema,
-			proctimeAttribute,
-			rowtimeAttributeDescriptors,
-			Optional.of(fieldMapping),
-			topic,
-			properties,
-			deserializationSchema,
-			startupMode,
-			specificStartupOffsets
-		);
-	}
-
-	@Override
-	protected KafkaTableSinkBase getExpectedKafkaTableSink(
-			TableSchema schema,
-			String topic,
-			Properties properties,
-			Optional<FlinkKafkaPartitioner<Row>> partitioner,
-			SerializationSchema<Row> serializationSchema) {
-
-		return new Kafka011TableSink(
-			schema,
-			topic,
-			properties,
-			partitioner,
-			serializationSchema
-		);
-	}
-}
+///*
+// * 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.api.common.serialization.SerializationSchema;
+//import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
+//import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
+//import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
+//import org.apache.flink.table.api.TableSchema;
+//import org.apache.flink.table.descriptors.KafkaValidator;
+//import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
+//import org.apache.flink.types.Row;
+//
+//import java.util.List;
+//import java.util.Map;
+//import java.util.Optional;
+//import java.util.Properties;
+//
+///**
+// * Test for {@link Kafka011TableSource} and {@link Kafka011TableSink} created
+// * by {@link Kafka011TableSourceSinkFactory}.
+// */
+//public class Kafka011TableSourceSinkFactoryTest extends KafkaTableSourceSinkFactoryTestBase {
+//
+//	@Override
+//	protected String getKafkaVersion() {
+//		return KafkaValidator.CONNECTOR_VERSION_VALUE_011;
+//	}
+//
+//	@Override
+//	@SuppressWarnings("unchecked")
+//	protected Class<FlinkKafkaConsumerBase<Row>> getExpectedFlinkKafkaConsumer() {
+//		return (Class) FlinkKafkaConsumer011.class;
+//	}
+//
+//	@Override
+//	protected Class<?> getExpectedFlinkKafkaProducer() {
+//		return FlinkKafkaProducer011.class;
+//	}
+//
+//	@Override
+//	protected KafkaTableSourceBase getExpectedKafkaTableSource(
+//			TableSchema schema,
+//			Optional<String> proctimeAttribute,
+//			List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors,
+//			Map<String, String> fieldMapping,
+//			String topic,
+//			Properties properties,
+//			DeserializationSchema<Row> deserializationSchema,
+//			StartupMode startupMode,
+//			Map<KafkaTopicPartition, Long> specificStartupOffsets) {
+//
+//		return new Kafka011TableSource(
+//			schema,
+//			proctimeAttribute,
+//			rowtimeAttributeDescriptors,
+//			Optional.of(fieldMapping),
+//			topic,
+//			properties,
+//			deserializationSchema,
+//			startupMode,
+//			specificStartupOffsets
+//		);
+//	}
+//
+//	@Override
+//	protected KafkaTableSinkBase getExpectedKafkaTableSink(
+//			TableSchema schema,
+//			String topic,
+//			Properties properties,
+//			Optional<FlinkKafkaPartitioner<Row>> partitioner,
+//			SerializationSchema<Row> serializationSchema) {
+//
+//		return new Kafka011TableSink(
+//			schema,
+//			topic,
+//			properties,
+//			partitioner,
+//			serializationSchema
+//		);
+//	}
+//}
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/internal/TransactionalIdsGeneratorTest.java b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/internal/TransactionalIdsGeneratorTest.java
index 7ad377986ee..0105ef19cd9 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/internal/TransactionalIdsGeneratorTest.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/internal/TransactionalIdsGeneratorTest.java
@@ -1,79 +1,79 @@
-/*
- * 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.internal;
-
-import org.junit.Test;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-
-import static org.junit.Assert.assertEquals;
-
-/**
- * Tests for {@link TransactionalIdsGenerator}.
- */
-public class TransactionalIdsGeneratorTest {
-	private static final int POOL_SIZE = 3;
-	private static final int SAFE_SCALE_DOWN_FACTOR = 3;
-	private static final int SUBTASKS_COUNT = 5;
-
-	@Test
-	public void testGenerateIdsToUse() {
-		TransactionalIdsGenerator generator = new TransactionalIdsGenerator("test", 2, SUBTASKS_COUNT, POOL_SIZE, SAFE_SCALE_DOWN_FACTOR);
-
-		assertEquals(
-			new HashSet<>(Arrays.asList("test-42", "test-43", "test-44")),
-			generator.generateIdsToUse(36));
-	}
-
-	/**
-	 * Ids to abort and to use should never clash between subtasks.
-	 */
-	@Test
-	public void testGeneratedIdsDoNotClash() {
-		List<Set<String>> idsToAbort = new ArrayList<>();
-		List<Set<String>> idsToUse = new ArrayList<>();
-
-		for (int subtask = 0; subtask < SUBTASKS_COUNT; subtask++) {
-			TransactionalIdsGenerator generator = new TransactionalIdsGenerator("test", subtask, SUBTASKS_COUNT, POOL_SIZE, SAFE_SCALE_DOWN_FACTOR);
-			idsToUse.add(generator.generateIdsToUse(0));
-			idsToAbort.add(generator.generateIdsToAbort());
-		}
-
-		for (int subtask1 = 0; subtask1 < SUBTASKS_COUNT; subtask1++) {
-			for (int subtask2 = 0; subtask2 < SUBTASKS_COUNT; subtask2++) {
-				if (subtask2 == subtask1) {
-					continue;
-				}
-				assertDisjoint(idsToAbort.get(subtask2), idsToAbort.get(subtask1));
-				assertDisjoint(idsToUse.get(subtask2), idsToUse.get(subtask1));
-				assertDisjoint(idsToAbort.get(subtask2), idsToUse.get(subtask1));
-			}
-		}
-	}
-
-	private <T> void assertDisjoint(Set<T> first, Set<T> second) {
-		HashSet<T> actual = new HashSet<>(first);
-		actual.retainAll(second);
-		assertEquals(Collections.emptySet(), actual);
-	}
-}
+///*
+// * 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.internal;
+//
+//import org.junit.Test;
+//
+//import java.util.ArrayList;
+//import java.util.Arrays;
+//import java.util.Collections;
+//import java.util.HashSet;
+//import java.util.List;
+//import java.util.Set;
+//
+//import static org.junit.Assert.assertEquals;
+//
+///**
+// * Tests for {@link TransactionalIdsGenerator}.
+// */
+//public class TransactionalIdsGeneratorTest {
+//	private static final int POOL_SIZE = 3;
+//	private static final int SAFE_SCALE_DOWN_FACTOR = 3;
+//	private static final int SUBTASKS_COUNT = 5;
+//
+//	@Test
+//	public void testGenerateIdsToUse() {
+//		TransactionalIdsGenerator generator = new TransactionalIdsGenerator("test", 2, SUBTASKS_COUNT, POOL_SIZE, SAFE_SCALE_DOWN_FACTOR);
+//
+//		assertEquals(
+//			new HashSet<>(Arrays.asList("test-42", "test-43", "test-44")),
+//			generator.generateIdsToUse(36));
+//	}
+//
+//	/**
+//	 * Ids to abort and to use should never clash between subtasks.
+//	 */
+//	@Test
+//	public void testGeneratedIdsDoNotClash() {
+//		List<Set<String>> idsToAbort = new ArrayList<>();
+//		List<Set<String>> idsToUse = new ArrayList<>();
+//
+//		for (int subtask = 0; subtask < SUBTASKS_COUNT; subtask++) {
+//			TransactionalIdsGenerator generator = new TransactionalIdsGenerator("test", subtask, SUBTASKS_COUNT, POOL_SIZE, SAFE_SCALE_DOWN_FACTOR);
+//			idsToUse.add(generator.generateIdsToUse(0));
+//			idsToAbort.add(generator.generateIdsToAbort());
+//		}
+//
+//		for (int subtask1 = 0; subtask1 < SUBTASKS_COUNT; subtask1++) {
+//			for (int subtask2 = 0; subtask2 < SUBTASKS_COUNT; subtask2++) {
+//				if (subtask2 == subtask1) {
+//					continue;
+//				}
+//				assertDisjoint(idsToAbort.get(subtask2), idsToAbort.get(subtask1));
+//				assertDisjoint(idsToUse.get(subtask2), idsToUse.get(subtask1));
+//				assertDisjoint(idsToAbort.get(subtask2), idsToUse.get(subtask1));
+//			}
+//		}
+//	}
+//
+//	private <T> void assertDisjoint(Set<T> first, Set<T> second) {
+//		HashSet<T> actual = new HashSet<>(first);
+//		actual.retainAll(second);
+//		assertEquals(Collections.emptySet(), actual);
+//	}
+//}
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
index 4ce8e62ca32..92f05089d70 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
@@ -168,6 +168,7 @@ public void runFailOnNoBrokerTest() throws Exception {
 			properties.setProperty("zookeeper.connect", "localhost:80");
 			properties.setProperty("group.id", "test");
 			properties.setProperty("request.timeout.ms", "3000"); // let the test fail fast
+			properties.setProperty("default.api.timeout.ms", "3000"); // KafkaProducer#partitionsFor
 			properties.setProperty("socket.timeout.ms", "3000");
 			properties.setProperty("session.timeout.ms", "2000");
 			properties.setProperty("fetch.max.wait.ms", "2000");
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaITCase.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaITCase.java
index 11aee4a65fd..3c17c3b3908 100644
--- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaITCase.java
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaITCase.java
@@ -63,7 +63,7 @@ public static void prepare() throws ClassNotFoundException {
 	//  Suite of Tests
 	// ------------------------------------------------------------------------
 
-	@Test(timeout = 120000)
+	@Test(timeout = 60000)
 	public void testFailOnNoBroker() throws Exception {
 		runFailOnNoBrokerTest();
 	}


 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services