You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2019/06/28 07:30:02 UTC
[flink] branch master updated (a57ba1c -> af0059c)
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from a57ba1c [hotfix][table-common] Add missing TIME type family
new 96dcf63 [hotfix] Remove unused method from Kafka Test Environments
new d43e6b8 [FLINK-11693] Add KafkaSerializationSchema that uses ProducerRecord
new d9c71e3 [FLINK-11693] Support FlinkKafkaPartitioner functionality with new KafkaSerializationSchema
new af0059c [FLINK-11693] Deprecate old constructors in modern KafkaProducer
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../streaming/connectors/kafka/Kafka010ITCase.java | 2 +-
.../connectors/kafka/KafkaTestEnvironmentImpl.java | 8 -
.../streaming/connectors/kafka/Kafka011ITCase.java | 2 +-
.../connectors/kafka/KafkaTestEnvironmentImpl.java | 16 --
.../streaming/connectors/kafka/Kafka08ITCase.java | 2 +-
.../connectors/kafka/KafkaTestEnvironmentImpl.java | 5 -
.../streaming/connectors/kafka/Kafka09ITCase.java | 2 +-
.../connectors/kafka/Kafka09SecuredRunITCase.java | 2 +-
.../connectors/kafka/KafkaTestEnvironmentImpl.java | 5 -
.../connectors/kafka/KafkaContextAware.java | 61 +++++
.../kafka/KafkaSerializationSchema.java} | 46 ++--
.../serialization/KeyedSerializationSchema.java | 4 +
.../connectors/kafka/KafkaConsumerTestBase.java | 63 +++++-
.../connectors/kafka/KafkaTestEnvironment.java | 9 +-
.../connectors/kafka/FlinkKafkaProducer.java | 251 +++++++++++++++++----
.../streaming/connectors/kafka/KafkaITCase.java | 9 +-
.../connectors/kafka/KafkaTestEnvironmentImpl.java | 19 +-
17 files changed, 379 insertions(+), 127 deletions(-)
create mode 100644 flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaContextAware.java
copy flink-connectors/{flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafka011Exception.java => flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaSerializationSchema.java} (50%)
[flink] 02/04: [FLINK-11693] Add KafkaSerializationSchema that uses
ProducerRecord
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d43e6b82ea16f49aee5006de8d63212af7fea8fc
Author: Aljoscha Krettek <al...@apache.org>
AuthorDate: Fri Feb 22 15:57:07 2019 +0100
[FLINK-11693] Add KafkaSerializationSchema that uses ProducerRecord
---
.../streaming/connectors/kafka/Kafka010ITCase.java | 2 +-
.../streaming/connectors/kafka/Kafka011ITCase.java | 2 +-
.../connectors/kafka/KafkaTestEnvironmentImpl.java | 1 -
.../streaming/connectors/kafka/Kafka08ITCase.java | 2 +-
.../streaming/connectors/kafka/Kafka09ITCase.java | 2 +-
.../connectors/kafka/Kafka09SecuredRunITCase.java | 2 +-
.../connectors/kafka/KafkaSerializationSchema.java | 45 +++++
.../serialization/KeyedSerializationSchema.java | 4 +
.../connectors/kafka/KafkaConsumerTestBase.java | 63 ++++++-
.../connectors/kafka/KafkaTestEnvironment.java | 5 +
.../connectors/kafka/FlinkKafkaProducer.java | 197 +++++++++++++++++----
.../streaming/connectors/kafka/KafkaITCase.java | 9 +-
.../connectors/kafka/KafkaTestEnvironmentImpl.java | 10 +-
13 files changed, 288 insertions(+), 56 deletions(-)
diff --git a/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010ITCase.java b/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010ITCase.java
index 6c1d1e2..0c8bdbb 100644
--- a/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010ITCase.java
+++ b/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010ITCase.java
@@ -115,7 +115,7 @@ public class Kafka010ITCase extends KafkaConsumerTestBase {
@Test(timeout = 60000)
public void testMultipleTopics() throws Exception {
- runProduceConsumeMultipleTopics();
+ runProduceConsumeMultipleTopics(true);
}
@Test(timeout = 60000)
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 3677daa..0a68766 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
@@ -123,7 +123,7 @@ public class Kafka011ITCase extends KafkaConsumerTestBase {
@Test(timeout = 60000)
public void testMultipleTopics() throws Exception {
- runProduceConsumeMultipleTopics();
+ runProduceConsumeMultipleTopics(true);
}
@Test(timeout = 60000)
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
index 23cd57e..a3982ba 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
@@ -21,7 +21,6 @@ import org.apache.flink.networking.NetworkFailuresProxy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.operators.StreamSink;
-import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkFixedPartitioner;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.connectors.kafka.testutils.ZooKeeperStringSerializer;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
diff --git a/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08ITCase.java b/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08ITCase.java
index e360c19..b97b3e9 100644
--- a/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08ITCase.java
+++ b/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08ITCase.java
@@ -237,7 +237,7 @@ public class Kafka08ITCase extends KafkaConsumerTestBase {
@Test(timeout = 60000)
public void testMultipleTopics() throws Exception {
- runProduceConsumeMultipleTopics();
+ runProduceConsumeMultipleTopics(true);
}
@Test(timeout = 60000)
diff --git a/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09ITCase.java b/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09ITCase.java
index 47d910d..8fdbbe7 100644
--- a/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09ITCase.java
+++ b/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09ITCase.java
@@ -109,7 +109,7 @@ public class Kafka09ITCase extends KafkaConsumerTestBase {
@Test(timeout = 60000)
public void testMultipleTopics() throws Exception {
- runProduceConsumeMultipleTopics();
+ runProduceConsumeMultipleTopics(true);
}
@Test(timeout = 60000)
diff --git a/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09SecuredRunITCase.java b/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09SecuredRunITCase.java
index 8cd61cc..f8ce2f5 100644
--- a/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09SecuredRunITCase.java
+++ b/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka09SecuredRunITCase.java
@@ -54,7 +54,7 @@ public class Kafka09SecuredRunITCase extends KafkaConsumerTestBase {
//The timeout for the test case is 2 times timeout of ZK connection
@Test(timeout = 600000)
public void testMultipleTopics() throws Exception {
- runProduceConsumeMultipleTopics();
+ runProduceConsumeMultipleTopics(true);
}
}
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaSerializationSchema.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaSerializationSchema.java
new file mode 100644
index 0000000..31fda0e
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaSerializationSchema.java
@@ -0,0 +1,45 @@
+/*
+ * 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.annotation.PublicEvolving;
+
+import org.apache.kafka.clients.producer.ProducerRecord;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+
+/**
+ * A {@link KafkaSerializationSchema} defines how to serialize values of type {@code T} into
+ * {@link ProducerRecord ProducerRecords}.
+ *
+ * @param <T> the type of values being serialized
+ */
+@PublicEvolving
+public interface KafkaSerializationSchema<T> extends Serializable {
+
+ /**
+ * Serializes given element and returns it as a {@link ProducerRecord}.
+ *
+ * @param element element to be serialized
+ * @param timestamp timestamp (can be null)
+ * @return Kafka {@link ProducerRecord}
+ */
+ ProducerRecord<byte[], byte[]> serialize(T element, @Nullable Long timestamp);
+}
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/KeyedSerializationSchema.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/KeyedSerializationSchema.java
index 2f610c2..b254f15 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/KeyedSerializationSchema.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/util/serialization/KeyedSerializationSchema.java
@@ -18,6 +18,7 @@
package org.apache.flink.streaming.util.serialization;
import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import java.io.Serializable;
@@ -27,7 +28,10 @@ import java.io.Serializable;
* to them in a specific format (for example as byte strings).
*
* @param <T> The type to be serialized.
+ *
+ * @deprecated Use {@link KafkaSerializationSchema}.
*/
+@Deprecated
@PublicEvolving
public interface KeyedSerializationSchema<T> extends Serializable {
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 8ca15b0..1582922 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
@@ -81,11 +81,13 @@ import kafka.server.KafkaServer;
import org.apache.commons.io.output.ByteArrayOutputStream;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.errors.TimeoutException;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Rule;
+import javax.annotation.Nullable;
import javax.management.MBeanServer;
import javax.management.ObjectName;
@@ -1128,7 +1130,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBaseWithFlink {
* Test producing and consuming into multiple topics.
* @throws Exception
*/
- public void runProduceConsumeMultipleTopics() throws Exception {
+ public void runProduceConsumeMultipleTopics(boolean useLegacySchema) throws Exception {
final int numTopics = 5;
final int numElements = 20;
@@ -1166,12 +1168,17 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBaseWithFlink {
}
});
- Tuple2WithTopicSchema schema = new Tuple2WithTopicSchema(env.getConfig());
-
Properties props = new Properties();
props.putAll(standardProps);
props.putAll(secureProps);
- kafkaServer.produceIntoKafka(stream, "dummy", schema, props, null);
+
+ if (useLegacySchema) {
+ Tuple2WithTopicSchema schema = new Tuple2WithTopicSchema(env.getConfig());
+ kafkaServer.produceIntoKafka(stream, "dummy", schema, props, null);
+ } else {
+ TestDeSerializer schema = new TestDeSerializer(env.getConfig());
+ kafkaServer.produceIntoKafka(stream, "dummy", schema, props);
+ }
env.execute("Write to topics");
@@ -1179,7 +1186,13 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBaseWithFlink {
env = StreamExecutionEnvironment.getExecutionEnvironment();
env.getConfig().disableSysoutLogging();
- stream = env.addSource(kafkaServer.getConsumer(topics, schema, props));
+ if (useLegacySchema) {
+ Tuple2WithTopicSchema schema = new Tuple2WithTopicSchema(env.getConfig());
+ stream = env.addSource(kafkaServer.getConsumer(topics, schema, props));
+ } else {
+ TestDeSerializer schema = new TestDeSerializer(env.getConfig());
+ stream = env.addSource(kafkaServer.getConsumer(topics, schema, props));
+ }
stream.flatMap(new FlatMapFunction<Tuple3<Integer, Integer, String>, Integer>() {
Map<String, Integer> countPerTopic = new HashMap<>(numTopics);
@@ -2194,12 +2207,12 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBaseWithFlink {
}
}
- private static class Tuple2WithTopicSchema implements KafkaDeserializationSchema<Tuple3<Integer, Integer, String>>,
- KeyedSerializationSchema<Tuple3<Integer, Integer, String>> {
+ private abstract static class TestDeserializer implements
+ KafkaDeserializationSchema<Tuple3<Integer, Integer, String>> {
- private final TypeSerializer<Tuple2<Integer, Integer>> ts;
+ protected final TypeSerializer<Tuple2<Integer, Integer>> ts;
- public Tuple2WithTopicSchema(ExecutionConfig ec) {
+ public TestDeserializer(ExecutionConfig ec) {
ts = TypeInformation.of(new TypeHint<Tuple2<Integer, Integer>>(){}).createSerializer(ec);
}
@@ -2219,6 +2232,14 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBaseWithFlink {
public TypeInformation<Tuple3<Integer, Integer, String>> getProducedType() {
return TypeInformation.of(new TypeHint<Tuple3<Integer, Integer, String>>(){});
}
+ }
+
+ private static class Tuple2WithTopicSchema extends TestDeserializer
+ implements KeyedSerializationSchema<Tuple3<Integer, Integer, String>> {
+
+ public Tuple2WithTopicSchema(ExecutionConfig ec) {
+ super(ec);
+ }
@Override
public byte[] serializeKey(Tuple3<Integer, Integer, String> element) {
@@ -2242,4 +2263,28 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBaseWithFlink {
return element.f2;
}
}
+
+ private static class TestDeSerializer extends TestDeserializer
+ implements KafkaSerializationSchema<Tuple3<Integer, Integer, String>> {
+
+ public TestDeSerializer(ExecutionConfig ec) {
+ super(ec);
+ }
+
+ @Override
+ public ProducerRecord<byte[], byte[]> serialize(
+ Tuple3<Integer, Integer, String> element, @Nullable Long timestamp) {
+ ByteArrayOutputStream by = new ByteArrayOutputStream();
+ DataOutputView out = new DataOutputViewStreamWrapper(by);
+ try {
+ ts.serialize(new Tuple2<>(element.f0, element.f1), out);
+ } catch (IOException e) {
+ throw new RuntimeException("Error" , e);
+ }
+ byte[] serializedValue = by.toByteArray();
+
+ return new ProducerRecord<>(element.f2, serializedValue);
+ }
+
+ }
}
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironment.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironment.java
index 336372d..3682280 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironment.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironment.java
@@ -154,6 +154,11 @@ public abstract class KafkaTestEnvironment {
KeyedSerializationSchema<T> serSchema, Properties props,
FlinkKafkaPartitioner<T> partitioner);
+ public <T> DataStreamSink<T> produceIntoKafka(DataStream<T> stream, String topic,
+ KafkaSerializationSchema<T> serSchema, Properties props) {
+ throw new RuntimeException("KafkaSerializationSchema is only supported on the modern Kafka Connector.");
+ }
+
// -- offset handlers
/**
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
index c7375eb..a8bff7c 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
@@ -216,11 +216,20 @@ public class FlinkKafkaProducer<IN>
* (Serializable) SerializationSchema for turning objects used with Flink into.
* byte[] for Kafka.
*/
- private final KeyedSerializationSchema<IN> schema;
+ @Nullable
+ private final KeyedSerializationSchema<IN> keyedSchema;
+
+ /**
+ * (Serializable) serialization schema for serializing records to
+ * {@link ProducerRecord ProducerRecords}.
+ */
+ @Nullable
+ private final KafkaSerializationSchema<IN> kafkaSchema;
/**
* User-provided partitioner for assigning an object to a Kafka partition for each topic.
*/
+ @Nullable
private final FlinkKafkaPartitioner<IN> flinkKafkaPartitioner;
/**
@@ -404,9 +413,6 @@ public class FlinkKafkaProducer<IN>
* partition (i.e. all records received by a sink subtask will end up in the same
* Kafka partition).
*
- * <p>To use a custom partitioner, please use
- * {@link #FlinkKafkaProducer(String, KeyedSerializationSchema, Properties, Optional, FlinkKafkaProducer.Semantic, int)} instead.
- *
* @param topicId
* ID of the Kafka topic.
* @param serializationSchema
@@ -483,25 +489,135 @@ public class FlinkKafkaProducer<IN>
* @param kafkaProducersPoolSize Overwrite default KafkaProducers pool size (see {@link FlinkKafkaProducer.Semantic#EXACTLY_ONCE}).
*/
public FlinkKafkaProducer(
- String defaultTopicId,
- KeyedSerializationSchema<IN> serializationSchema,
+ String defaultTopicId,
+ KeyedSerializationSchema<IN> serializationSchema,
+ Properties producerConfig,
+ Optional<FlinkKafkaPartitioner<IN>> customPartitioner,
+ FlinkKafkaProducer.Semantic semantic,
+ int kafkaProducersPoolSize) {
+ this(
+ defaultTopicId,
+ serializationSchema,
+ customPartitioner.orElse(null),
+ null, /* kafka serialization schema */
+ producerConfig,
+ semantic,
+ kafkaProducersPoolSize);
+ }
+
+ /**
+ * Creates a {@link FlinkKafkaProducer} for a given topic. The sink produces its input to
+ * the topic. It accepts a {@link KafkaSerializationSchema} for serializing records to
+ * a {@link ProducerRecord}, including partitioning information.
+ *
+ * @param defaultTopic The default topic to write data to
+ * @param serializationSchema A serializable serialization schema for turning user objects into a kafka-consumable byte[] supporting key/value messages
+ * @param producerConfig Configuration properties for the KafkaProducer. 'bootstrap.servers.' is the only required argument.
+ * @param semantic Defines semantic that will be used by this producer (see {@link FlinkKafkaProducer.Semantic}).
+ */
+ public FlinkKafkaProducer(
+ String defaultTopic,
+ KafkaSerializationSchema<IN> serializationSchema,
+ Properties producerConfig,
+ FlinkKafkaProducer.Semantic semantic) {
+ this(
+ defaultTopic,
+ serializationSchema,
+ producerConfig,
+ semantic,
+ DEFAULT_KAFKA_PRODUCERS_POOL_SIZE);
+ }
+
+ /**
+ * Creates a FlinkKafkaProducer for a given topic. The sink produces its input to
+ * the topic. It accepts a {@link KafkaSerializationSchema} and possibly a custom {@link FlinkKafkaPartitioner}.
+ *
+ * @param defaultTopic The default topic to write data to
+ * @param serializationSchema A serializable serialization schema for turning user objects into a kafka-consumable byte[] supporting key/value messages
+ * @param producerConfig Configuration properties for the KafkaProducer. 'bootstrap.servers.' is the only required argument.
+ * @param semantic Defines semantic that will be used by this producer (see {@link FlinkKafkaProducer.Semantic}).
+ * @param kafkaProducersPoolSize Overwrite default KafkaProducers pool size (see {@link FlinkKafkaProducer.Semantic#EXACTLY_ONCE}).
+ */
+ public FlinkKafkaProducer(
+ String defaultTopic,
+ KafkaSerializationSchema<IN> serializationSchema,
Properties producerConfig,
- Optional<FlinkKafkaPartitioner<IN>> customPartitioner,
FlinkKafkaProducer.Semantic semantic,
int kafkaProducersPoolSize) {
+ this(
+ defaultTopic,
+ null, null, /* keyed schema and FlinkKafkaPartitioner */
+ serializationSchema,
+ producerConfig,
+ semantic,
+ kafkaProducersPoolSize);
+ }
+
+ /**
+ * Creates a FlinkKafkaProducer for a given topic. The sink produces its input to
+ * the topic. It accepts a {@link KafkaSerializationSchema} and possibly a custom {@link FlinkKafkaPartitioner}.
+ *
+ * <p>If a partitioner is not provided, written records will be partitioned by the attached key of each
+ * record (as determined by {@link KeyedSerializationSchema#serializeKey(Object)}). If written records do not
+ * have a key (i.e., {@link KeyedSerializationSchema#serializeKey(Object)} returns {@code null}), they
+ * will be distributed to Kafka partitions in a round-robin fashion.
+ *
+ * @param defaultTopic The default topic to write data to
+ * @param keyedSchema A serializable serialization schema for turning user objects into a kafka-consumable byte[] supporting key/value messages
+ * @param customPartitioner A serializable partitioner for assigning messages to Kafka partitions.
+ * If a partitioner is not provided, records will be partitioned by the key of each record
+ * (determined by {@link KeyedSerializationSchema#serializeKey(Object)}). If the keys
+ * are {@code null}, then records will be distributed to Kafka partitions in a
+ * round-robin fashion.
+ * @param kafkaSchema A serializable serialization schema for turning user objects into a kafka-consumable byte[] supporting key/value messages
+ * @param producerConfig Configuration properties for the KafkaProducer. 'bootstrap.servers.' is the only required argument.
+ * @param semantic Defines semantic that will be used by this producer (see {@link FlinkKafkaProducer.Semantic}).
+ * @param kafkaProducersPoolSize Overwrite default KafkaProducers pool size (see {@link FlinkKafkaProducer.Semantic#EXACTLY_ONCE}).
+ */
+ private FlinkKafkaProducer(
+ String defaultTopic,
+ KeyedSerializationSchema<IN> keyedSchema,
+ FlinkKafkaPartitioner<IN> customPartitioner,
+ KafkaSerializationSchema<IN> kafkaSchema,
+ Properties producerConfig,
+ FlinkKafkaProducer.Semantic semantic,
+ int kafkaProducersPoolSize) {
super(new FlinkKafkaProducer.TransactionStateSerializer(), new FlinkKafkaProducer.ContextStateSerializer());
- this.defaultTopicId = checkNotNull(defaultTopicId, "defaultTopicId is null");
- this.schema = checkNotNull(serializationSchema, "serializationSchema is null");
+ this.defaultTopicId = checkNotNull(defaultTopic, "defaultTopic is null");
+
+ if (kafkaSchema != null) {
+ this.keyedSchema = null;
+ this.kafkaSchema = kafkaSchema;
+ this.flinkKafkaPartitioner = null;
+ ClosureCleaner.clean(
+ this.kafkaSchema, ExecutionConfig.ClosureCleanerLevel.RECURSIVE, true);
+
+ if (customPartitioner != null) {
+ throw new IllegalArgumentException("Customer partitioner can only be used when" +
+ "using a KeyedSerializationSchema or SerializationSchema.");
+ }
+ } else if (keyedSchema != null) {
+ this.kafkaSchema = null;
+ this.keyedSchema = keyedSchema;
+ this.flinkKafkaPartitioner = customPartitioner;
+ ClosureCleaner.clean(
+ this.flinkKafkaPartitioner,
+ ExecutionConfig.ClosureCleanerLevel.RECURSIVE,
+ true);
+ ClosureCleaner.clean(
+ this.keyedSchema, ExecutionConfig.ClosureCleanerLevel.RECURSIVE, true);
+ } else {
+ throw new IllegalArgumentException(
+ "You must provide either a KafkaSerializationSchema or a" +
+ "KeyedSerializationSchema.");
+ }
+
this.producerConfig = checkNotNull(producerConfig, "producerConfig is null");
- this.flinkKafkaPartitioner = checkNotNull(customPartitioner, "customPartitioner is null").orElse(null);
this.semantic = checkNotNull(semantic, "semantic is null");
this.kafkaProducersPoolSize = kafkaProducersPoolSize;
checkState(kafkaProducersPoolSize > 0, "kafkaProducersPoolSize must be non empty");
- ClosureCleaner.clean(this.flinkKafkaPartitioner, ExecutionConfig.ClosureCleanerLevel.RECURSIVE, true);
- ClosureCleaner.ensureSerializable(serializationSchema);
-
// set the producer configuration properties for kafka record key value serializers.
if (!producerConfig.containsKey(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG)) {
this.producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
@@ -539,7 +655,7 @@ public class FlinkKafkaProducer<IN>
transactionTimeout = ((Number) object).longValue();
} else {
throw new IllegalArgumentException(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG
- + " must be numeric, was " + object);
+ + " must be numeric, was " + object);
}
super.setTransactionTimeout(transactionTimeout);
super.enableTransactionTimeoutWarnings(0.8);
@@ -625,34 +741,39 @@ public class FlinkKafkaProducer<IN>
public void invoke(FlinkKafkaProducer.KafkaTransactionState transaction, IN next, Context context) throws FlinkKafkaException {
checkErroneous();
- byte[] serializedKey = schema.serializeKey(next);
- byte[] serializedValue = schema.serializeValue(next);
- String targetTopic = schema.getTargetTopic(next);
- if (targetTopic == null) {
- targetTopic = defaultTopicId;
- }
+ ProducerRecord<byte[], byte[]> record;
+ if (keyedSchema != null) {
+ byte[] serializedKey = keyedSchema.serializeKey(next);
+ byte[] serializedValue = keyedSchema.serializeValue(next);
+ String targetTopic = keyedSchema.getTargetTopic(next);
+ if (targetTopic == null) {
+ targetTopic = defaultTopicId;
+ }
- Long timestamp = null;
- if (this.writeTimestampToKafka) {
- timestamp = context.timestamp();
- }
+ Long timestamp = null;
+ if (this.writeTimestampToKafka) {
+ timestamp = context.timestamp();
+ }
- ProducerRecord<byte[], byte[]> record;
- int[] partitions = topicPartitionsMap.get(targetTopic);
- if (null == partitions) {
- partitions = getPartitionsByTopic(targetTopic, transaction.producer);
- topicPartitionsMap.put(targetTopic, partitions);
- }
- if (flinkKafkaPartitioner != null) {
- record = new ProducerRecord<>(
- targetTopic,
- flinkKafkaPartitioner.partition(next, serializedKey, serializedValue, targetTopic, partitions),
- timestamp,
- serializedKey,
- serializedValue);
+ int[] partitions = topicPartitionsMap.get(targetTopic);
+ if (null == partitions) {
+ partitions = getPartitionsByTopic(targetTopic, transaction.producer);
+ topicPartitionsMap.put(targetTopic, partitions);
+ }
+ if (flinkKafkaPartitioner != null) {
+ record = new ProducerRecord<>(
+ targetTopic,
+ flinkKafkaPartitioner.partition(next, serializedKey, serializedValue, targetTopic, partitions),
+ timestamp,
+ serializedKey,
+ serializedValue);
+ } else {
+ record = new ProducerRecord<>(targetTopic, null, timestamp, serializedKey, serializedValue);
+ }
} else {
- record = new ProducerRecord<>(targetTopic, null, timestamp, serializedKey, serializedValue);
+ record = kafkaSchema.serialize(next, context.timestamp());
}
+
pendingRecords.incrementAndGet();
transaction.producer.send(record, callback);
}
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 8729319..d79d509 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
@@ -122,8 +122,13 @@ public class KafkaITCase extends KafkaConsumerTestBase {
}
@Test(timeout = 60000)
- public void testMultipleTopics() throws Exception {
- runProduceConsumeMultipleTopics();
+ public void testMultipleTopicsWithLegacySerializer() throws Exception {
+ runProduceConsumeMultipleTopics(true);
+ }
+
+ @Test(timeout = 60000)
+ public void testMultipleTopicsWithKafkaSerializer() throws Exception {
+ runProduceConsumeMultipleTopics(false);
}
@Test(timeout = 60000)
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
index 710c753..d5c955e 100644
--- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
@@ -21,7 +21,6 @@ import org.apache.flink.networking.NetworkFailuresProxy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.operators.StreamSink;
-import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkFixedPartitioner;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.flink.util.NetUtils;
@@ -274,6 +273,15 @@ public class KafkaTestEnvironmentImpl extends KafkaTestEnvironment {
}
@Override
+ public <T> DataStreamSink<T> produceIntoKafka(DataStream<T> stream, String topic, KafkaSerializationSchema<T> serSchema, Properties props) {
+ return stream.addSink(new FlinkKafkaProducer<T>(
+ topic,
+ serSchema,
+ props,
+ producerSemantic));
+ }
+
+ @Override
public KafkaOffsetHandler createOffsetHandler() {
return new KafkaOffsetHandlerImpl();
}
[flink] 04/04: [FLINK-11693] Deprecate old constructors in modern
KafkaProducer
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit af0059ce43b667687d25e0d1c5a5b4f3f4dbbb08
Author: Aljoscha Krettek <al...@apache.org>
AuthorDate: Wed Jun 19 12:13:33 2019 +0200
[FLINK-11693] Deprecate old constructors in modern KafkaProducer
---
.../connectors/kafka/FlinkKafkaProducer.java | 24 ++++++++++++++++++++++
1 file changed, 24 insertions(+)
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
index e2c8bbc..6314b58 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
@@ -288,7 +288,10 @@ public class FlinkKafkaProducer<IN>
* ID of the Kafka topic.
* @param serializationSchema
* User defined (keyless) serialization schema.
+ *
+ * @deprecated use {@link #FlinkKafkaProducer(String, KafkaSerializationSchema, Properties, FlinkKafkaProducer.Semantic)}
*/
+ @Deprecated
public FlinkKafkaProducer(String brokerList, String topicId, SerializationSchema<IN> serializationSchema) {
this(
topicId,
@@ -315,7 +318,10 @@ public class FlinkKafkaProducer<IN>
* User defined key-less serialization schema.
* @param producerConfig
* Properties with the producer configuration.
+ *
+ * @deprecated use {@link #FlinkKafkaProducer(String, KafkaSerializationSchema, Properties, FlinkKafkaProducer.Semantic)}
*/
+ @Deprecated
public FlinkKafkaProducer(String topicId, SerializationSchema<IN> serializationSchema, Properties producerConfig) {
this(
topicId,
@@ -338,7 +344,10 @@ public class FlinkKafkaProducer<IN>
* @param customPartitioner A serializable partitioner for assigning messages to Kafka partitions.
* If a partitioner is not provided, records will be distributed to Kafka partitions
* in a round-robin fashion.
+ *
+ * @deprecated use {@link #FlinkKafkaProducer(String, KafkaSerializationSchema, Properties, FlinkKafkaProducer.Semantic)}
*/
+ @Deprecated
public FlinkKafkaProducer(
String topicId,
SerializationSchema<IN> serializationSchema,
@@ -368,7 +377,10 @@ public class FlinkKafkaProducer<IN>
* ID of the Kafka topic.
* @param serializationSchema
* User defined serialization schema supporting key/value messages
+ *
+ * @deprecated use {@link #FlinkKafkaProducer(String, KafkaSerializationSchema, Properties, FlinkKafkaProducer.Semantic)}
*/
+ @Deprecated
public FlinkKafkaProducer(String brokerList, String topicId, KeyedSerializationSchema<IN> serializationSchema) {
this(
topicId,
@@ -395,7 +407,10 @@ public class FlinkKafkaProducer<IN>
* User defined serialization schema supporting key/value messages
* @param producerConfig
* Properties with the producer configuration.
+ *
+ * @deprecated use {@link #FlinkKafkaProducer(String, KafkaSerializationSchema, Properties, FlinkKafkaProducer.Semantic)}
*/
+ @Deprecated
public FlinkKafkaProducer(String topicId, KeyedSerializationSchema<IN> serializationSchema, Properties producerConfig) {
this(
topicId,
@@ -421,7 +436,10 @@ public class FlinkKafkaProducer<IN>
* Properties with the producer configuration.
* @param semantic
* Defines semantic that will be used by this producer (see {@link FlinkKafkaProducer.Semantic}).
+ *
+ * @deprecated use {@link #FlinkKafkaProducer(String, KafkaSerializationSchema, Properties, FlinkKafkaProducer.Semantic)}
*/
+ @Deprecated
public FlinkKafkaProducer(
String topicId,
KeyedSerializationSchema<IN> serializationSchema,
@@ -453,7 +471,10 @@ public class FlinkKafkaProducer<IN>
* (determined by {@link KeyedSerializationSchema#serializeKey(Object)}). If the keys
* are {@code null}, then records will be distributed to Kafka partitions in a
* round-robin fashion.
+ *
+ * @deprecated use {@link #FlinkKafkaProducer(String, KafkaSerializationSchema, Properties, FlinkKafkaProducer.Semantic)}
*/
+ @Deprecated
public FlinkKafkaProducer(
String defaultTopicId,
KeyedSerializationSchema<IN> serializationSchema,
@@ -487,7 +508,10 @@ public class FlinkKafkaProducer<IN>
* round-robin fashion.
* @param semantic Defines semantic that will be used by this producer (see {@link FlinkKafkaProducer.Semantic}).
* @param kafkaProducersPoolSize Overwrite default KafkaProducers pool size (see {@link FlinkKafkaProducer.Semantic#EXACTLY_ONCE}).
+ *
+ * @deprecated use {@link #FlinkKafkaProducer(String, KafkaSerializationSchema, Properties, FlinkKafkaProducer.Semantic)}
*/
+ @Deprecated
public FlinkKafkaProducer(
String defaultTopicId,
KeyedSerializationSchema<IN> serializationSchema,
[flink] 03/04: [FLINK-11693] Support FlinkKafkaPartitioner
functionality with new KafkaSerializationSchema
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d9c71e323d7765d03ecc0359692ec33d79abf426
Author: Aljoscha Krettek <al...@apache.org>
AuthorDate: Wed Jun 26 12:01:02 2019 +0200
[FLINK-11693] Support FlinkKafkaPartitioner functionality with new KafkaSerializationSchema
---
.../connectors/kafka/KafkaContextAware.java | 61 ++++++++++++++++++++++
.../connectors/kafka/KafkaSerializationSchema.java | 9 +++-
.../connectors/kafka/FlinkKafkaProducer.java | 32 +++++++++++-
3 files changed, 99 insertions(+), 3 deletions(-)
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaContextAware.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaContextAware.java
new file mode 100644
index 0000000..22399c0
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaContextAware.java
@@ -0,0 +1,61 @@
+/*
+ * 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.annotation.PublicEvolving;
+
+/**
+ * An interface for {@link KafkaSerializationSchema KafkaSerializationSchemas} that need information
+ * about the context where the Kafka Producer is running along with information about the available
+ * partitions.
+ *
+ * <p>You only need to override the methods for the information that you need. However, {@link
+ * #getTargetTopic(Object)} is required because it is used to determine the available partitions.
+ */
+@PublicEvolving
+public interface KafkaContextAware<T> {
+
+
+ /**
+ * Sets the number of the parallel subtask that the Kafka Producer is running on. The numbering
+ * starts from 0 and goes up to parallelism-1 (parallelism as returned by {@link
+ * #setNumParallelInstances(int)}).
+ */
+ default void setParallelInstanceId(int parallelInstanceId) {
+ }
+
+ /**
+ * Sets the parallelism with which the parallel task of the Kafka Producer runs.
+ */
+ default void setNumParallelInstances(int numParallelInstances) {
+ }
+
+ /**
+ * Sets the available partitions for the topic returned from {@link #getTargetTopic(Object)}.
+ */
+ default void setPartitions(int[] partitions) {
+ }
+
+ /**
+ * Returns the topic that the presented element should be sent to. This is not used for setting
+ * the topic (this is done via the {@link org.apache.kafka.clients.producer.ProducerRecord} that
+ * is returned from {@link KafkaSerializationSchema#serialize(Object, Long)}, it is only used
+ * for getting the available partitions that are presented to {@link #setPartitions(int[])}.
+ */
+ String getTargetTopic(T element);
+}
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaSerializationSchema.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaSerializationSchema.java
index 31fda0e..37d30fc 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaSerializationSchema.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaSerializationSchema.java
@@ -26,8 +26,13 @@ import javax.annotation.Nullable;
import java.io.Serializable;
/**
- * A {@link KafkaSerializationSchema} defines how to serialize values of type {@code T} into
- * {@link ProducerRecord ProducerRecords}.
+ * A {@link KafkaSerializationSchema} defines how to serialize values of type {@code T} into {@link
+ * ProducerRecord ProducerRecords}.
+ *
+ * <p>Please also implement {@link KafkaContextAware} if your serialization schema needs
+ * information
+ * about the available partitions and the number of parallel subtasks along with the subtask ID on
+ * which the Kafka Producer is running.
*
* @param <T> the type of values being serialized
*/
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
index a8bff7c..e2c8bbc 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
@@ -770,8 +770,30 @@ public class FlinkKafkaProducer<IN>
} else {
record = new ProducerRecord<>(targetTopic, null, timestamp, serializedKey, serializedValue);
}
- } else {
+ } else if (kafkaSchema != null) {
+ if (kafkaSchema instanceof KafkaContextAware) {
+ @SuppressWarnings("unchecked")
+ KafkaContextAware<IN> contextAwareSchema =
+ (KafkaContextAware<IN>) kafkaSchema;
+
+ String targetTopic = contextAwareSchema.getTargetTopic(next);
+ if (targetTopic == null) {
+ targetTopic = defaultTopicId;
+ }
+ int[] partitions = topicPartitionsMap.get(targetTopic);
+
+ if (null == partitions) {
+ partitions = getPartitionsByTopic(targetTopic, transaction.producer);
+ topicPartitionsMap.put(targetTopic, partitions);
+ }
+
+ contextAwareSchema.setPartitions(partitions);
+ }
record = kafkaSchema.serialize(next, context.timestamp());
+ } else {
+ throw new RuntimeException(
+ "We have neither KafkaSerializationSchema nor KeyedSerializationSchema, this" +
+ "is a bug.");
}
pendingRecords.incrementAndGet();
@@ -1101,6 +1123,14 @@ public class FlinkKafkaProducer<IN>
flinkKafkaPartitioner.open(ctx.getIndexOfThisSubtask(), ctx.getNumberOfParallelSubtasks());
}
+ if (kafkaSchema instanceof KafkaContextAware) {
+ KafkaContextAware<IN> contextAwareSchema =
+ (KafkaContextAware<IN>) kafkaSchema;
+
+ contextAwareSchema.setParallelInstanceId(ctx.getIndexOfThisSubtask());
+ contextAwareSchema.setNumParallelInstances(ctx.getNumberOfParallelSubtasks());
+ }
+
LOG.info("Starting FlinkKafkaInternalProducer ({}/{}) to produce into default topic {}",
ctx.getIndexOfThisSubtask() + 1, ctx.getNumberOfParallelSubtasks(), defaultTopicId);
[flink] 01/04: [hotfix] Remove unused method from Kafka Test
Environments
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 96dcf63f553a734940b5f76d161687c20efd3f91
Author: Aljoscha Krettek <al...@apache.org>
AuthorDate: Wed Jun 5 11:19:14 2019 +0200
[hotfix] Remove unused method from Kafka Test Environments
---
.../connectors/kafka/KafkaTestEnvironmentImpl.java | 8 --------
.../connectors/kafka/KafkaTestEnvironmentImpl.java | 15 ---------------
.../connectors/kafka/KafkaTestEnvironmentImpl.java | 5 -----
.../connectors/kafka/KafkaTestEnvironmentImpl.java | 5 -----
.../streaming/connectors/kafka/KafkaTestEnvironment.java | 6 ------
.../connectors/kafka/KafkaTestEnvironmentImpl.java | 15 ---------------
6 files changed, 54 deletions(-)
diff --git a/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java b/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
index 9e51aac..64649ee 100644
--- a/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
+++ b/flink-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
@@ -166,14 +166,6 @@ public class KafkaTestEnvironmentImpl extends KafkaTestEnvironment {
}
@Override
- public <T> DataStreamSink<T> writeToKafkaWithTimestamps(DataStream<T> stream, String topic, KeyedSerializationSchema<T> serSchema, Properties props) {
- FlinkKafkaProducer010<T> prod = new FlinkKafkaProducer010<>(topic, serSchema, props);
- prod.setFlushOnCheckpoint(true);
- prod.setWriteTimestampToKafka(true);
- return stream.addSink(prod);
- }
-
- @Override
public KafkaOffsetHandler createOffsetHandler() {
return new KafkaOffsetHandlerImpl();
}
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
index ae237f6..23cd57e 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
@@ -296,21 +296,6 @@ public class KafkaTestEnvironmentImpl extends KafkaTestEnvironment {
}
@Override
- public <T> DataStreamSink<T> writeToKafkaWithTimestamps(DataStream<T> stream, String topic, KeyedSerializationSchema<T> serSchema, Properties props) {
- FlinkKafkaProducer011<T> prod = new FlinkKafkaProducer011<>(
- topic,
- serSchema,
- props,
- Optional.of(new FlinkFixedPartitioner<>()),
- producerSemantic,
- FlinkKafkaProducer011.DEFAULT_KAFKA_PRODUCERS_POOL_SIZE);
-
- prod.setWriteTimestampToKafka(true);
-
- return stream.addSink(prod);
- }
-
- @Override
public KafkaOffsetHandler createOffsetHandler() {
return new KafkaOffsetHandlerImpl();
}
diff --git a/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java b/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
index 15c83e4..6836afb 100644
--- a/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
+++ b/flink-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
@@ -160,11 +160,6 @@ public class KafkaTestEnvironmentImpl extends KafkaTestEnvironment {
}
@Override
- public <T> DataStreamSink<T> writeToKafkaWithTimestamps(DataStream<T> stream, String topic, KeyedSerializationSchema<T> serSchema, Properties props) {
- throw new UnsupportedOperationException();
- }
-
- @Override
public KafkaOffsetHandler createOffsetHandler() {
return new KafkaOffsetHandlerImpl();
}
diff --git a/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java b/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
index 7a5c9e5..9d03319 100644
--- a/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
+++ b/flink-connectors/flink-connector-kafka-0.9/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
@@ -151,11 +151,6 @@ public class KafkaTestEnvironmentImpl extends KafkaTestEnvironment {
}
@Override
- public <T> DataStreamSink<T> writeToKafkaWithTimestamps(DataStream<T> stream, String topic, KeyedSerializationSchema<T> serSchema, Properties props) {
- throw new UnsupportedOperationException();
- }
-
- @Override
public KafkaOffsetHandler createOffsetHandler() {
return new KafkaOffsetHandlerImpl();
}
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironment.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironment.java
index a787f27..336372d 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironment.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironment.java
@@ -154,12 +154,6 @@ public abstract class KafkaTestEnvironment {
KeyedSerializationSchema<T> serSchema, Properties props,
FlinkKafkaPartitioner<T> partitioner);
- public abstract <T> DataStreamSink<T> writeToKafkaWithTimestamps(
- DataStream<T> stream,
- String topic,
- KeyedSerializationSchema<T> serSchema,
- Properties props);
-
// -- offset handlers
/**
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
index d6a7705..710c753 100644
--- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
@@ -274,21 +274,6 @@ public class KafkaTestEnvironmentImpl extends KafkaTestEnvironment {
}
@Override
- public <T> DataStreamSink<T> writeToKafkaWithTimestamps(DataStream<T> stream, String topic, KeyedSerializationSchema<T> serSchema, Properties props) {
- FlinkKafkaProducer<T> prod = new FlinkKafkaProducer<T>(
- topic,
- serSchema,
- props,
- Optional.of(new FlinkFixedPartitioner<>()),
- producerSemantic,
- FlinkKafkaProducer.DEFAULT_KAFKA_PRODUCERS_POOL_SIZE);
-
- prod.setWriteTimestampToKafka(true);
-
- return stream.addSink(prod);
- }
-
- @Override
public KafkaOffsetHandler createOffsetHandler() {
return new KafkaOffsetHandlerImpl();
}