From 4bd36b45ab53940e63dbbc98e21fe51ac4f7a3ad Mon Sep 17 00:00:00 2001 From: Sebastian Klemke Date: Mon, 4 Jan 2016 14:52:52 +0100 Subject: [PATCH 1/2] [FLINK-3188] Pass deletes to KeyedDeserializationSchema --- .../kafka/internals/LegacyFetcher.java | 17 ++-- .../kafka/KafkaConsumerTestBase.java | 80 ++++++++++++++++++- .../connectors/kafka/KafkaITCase.java | 9 ++- 3 files changed, 92 insertions(+), 14 deletions(-) diff --git a/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/LegacyFetcher.java b/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/LegacyFetcher.java index 4233c18d60ffe..b51ad6166924f 100644 --- a/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/LegacyFetcher.java +++ b/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/LegacyFetcher.java @@ -448,19 +448,18 @@ public void run() { final long offset = msg.offset(); - // put value into byte array ByteBuffer payload = msg.message().payload(); + + // If the message value is null, this represents a delete command for the message key. + // Log this and pass it on to the client who might want to also receive delete messages. + byte[] valueBytes; if (payload == null) { - // This message has no value (which means it has been deleted from the Kafka topic) deletedMessages++; - // advance offset in state to avoid re-reading the message - synchronized (sourceContext.getCheckpointLock()) { - offsetsState.put(topicPartition, offset); - } - continue; + valueBytes = null; + } else { + valueBytes = new byte[payload.remaining()]; + payload.get(valueBytes); } - byte[] valueBytes = new byte[payload.remaining()]; - payload.get(valueBytes); // put key into byte array byte[] keyBytes = null; diff --git a/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java b/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java index 4f7138452c7ba..d75a15c0aced3 100644 --- a/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java +++ b/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java @@ -41,12 +41,15 @@ import org.apache.flink.api.common.typeutils.TypeSerializer; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.api.java.tuple.Tuple3; +import org.apache.flink.api.java.typeutils.TupleTypeInfo; +import org.apache.flink.api.java.typeutils.TypeExtractor; import org.apache.flink.api.java.typeutils.TypeInfoParser; import org.apache.flink.api.java.typeutils.runtime.ByteArrayInputView; import org.apache.flink.client.program.ProgramInvocationException; import org.apache.flink.configuration.Configuration; import org.apache.flink.runtime.client.JobExecutionException; import org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException; +import org.apache.flink.runtime.util.DataOutputSerializer; import org.apache.flink.streaming.api.checkpoint.CheckpointNotifier; import org.apache.flink.streaming.api.checkpoint.Checkpointed; import org.apache.flink.streaming.api.datastream.DataStream; @@ -75,6 +78,7 @@ import org.apache.flink.streaming.util.serialization.SimpleStringSchema; import org.apache.flink.streaming.util.serialization.TypeInformationKeyValueSerializationSchema; import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema; +import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchemaWrapper; import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema; import org.apache.flink.streaming.util.serialization.TypeInformationSerializationSchema; import org.apache.flink.testutils.junit.RetryOnException; @@ -107,6 +111,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -122,10 +127,20 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase { // ------------------------------------------------------------------------ protected abstract FlinkKafkaConsumer getConsumer( - List topics, DeserializationSchema deserializationSchema, Properties props); + List topics, KeyedDeserializationSchema deserializationSchema, Properties props); + + protected FlinkKafkaConsumer getConsumer( + List topics, DeserializationSchema deserializationSchema, Properties props) { + return getConsumer(topics, new KeyedDeserializationSchemaWrapper(deserializationSchema), props); + } protected FlinkKafkaConsumer getConsumer( String topic, DeserializationSchema deserializationSchema, Properties props) { + return getConsumer(Collections.singletonList(topic), new KeyedDeserializationSchemaWrapper(deserializationSchema), props); + } + + protected FlinkKafkaConsumer getConsumer( + String topic, KeyedDeserializationSchema deserializationSchema, Properties props) { return getConsumer(Collections.singletonList(topic), deserializationSchema, props); } @@ -832,7 +847,7 @@ public void runConsumeMultipleTopics() throws java.lang.Exception { Assert.assertEquals((NUM_TOPICS * (NUM_TOPICS + 1))/2, topicPartitions.size()); KeyedDeserializationSchema> readSchema = new Tuple2WithTopicDeserializationSchema(env.getConfig()); - DataStreamSource> stream = env.addSource(new FlinkKafkaConsumer082<>(topics, readSchema, standardProps)); + DataStreamSource> stream = env.addSource(getConsumer(topics, readSchema, standardProps)); stream.flatMap(new FlatMapFunction, Integer>() { Map countPerTopic = new HashMap<>(NUM_TOPICS); @@ -1120,7 +1135,7 @@ public void cancel() { KeyedDeserializationSchema> readSchema = new TypeInformationKeyValueSerializationSchema<>(Long.class, PojoValue.class, env.getConfig()); - DataStream> fromKafka = env.addSource(new FlinkKafkaConsumer082<>(topic, readSchema, standardProps)); + DataStream> fromKafka = env.addSource(getConsumer(topic, readSchema, standardProps)); fromKafka.flatMap(new RichFlatMapFunction, Object>() { long counter = 0; @Override @@ -1152,7 +1167,66 @@ public static class PojoValue { public PojoValue() {} } + public void runAllDeletesTest() throws Exception { + final String topic = "alldeletestest"; + createTestTopic(topic, 1, 1); + final int ELEMENT_COUNT = 300; + + // ----------- Write some data into Kafka ------------------- + + StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment("localhost", flinkPort); + env.setParallelism(1); + env.setNumberOfExecutionRetries(3); + env.getConfig().disableSysoutLogging(); + + DataStream> kvStream = env.addSource(new SourceFunction>() { + @Override + public void run(SourceContext> ctx) throws Exception { + Random rnd = new Random(1337); + for (long i = 0; i < ELEMENT_COUNT; i++) { + final byte[] key = new byte[200]; + rnd.nextBytes(key); + ctx.collect(new Tuple2<>(key, (PojoValue) null)); + } + } + @Override + public void cancel() { + } + }); + + TypeInformationKeyValueSerializationSchema schema = new TypeInformationKeyValueSerializationSchema<>(byte[].class, PojoValue.class, env.getConfig()); + kvStream.addSink(new FlinkKafkaProducer<>(topic, schema, + FlinkKafkaProducer.getPropertiesFromBrokerList(brokerConnectionStrings))); + env.execute("Write deletes to Kafka"); + + // ----------- Read the data again ------------------- + + env = StreamExecutionEnvironment.createRemoteEnvironment("localhost", flinkPort); + env.setParallelism(1); + env.setNumberOfExecutionRetries(3); + env.getConfig().disableSysoutLogging(); + + DataStream> fromKafka = env.addSource(getConsumer(topic, schema, standardProps)); + + fromKafka.flatMap(new RichFlatMapFunction, Object>() { + long counter = 0; + @Override + public void flatMap(Tuple2 value, Collector out) throws Exception { + // ensure that deleted messages are passed as nulls + assertNull(value.f1); + counter++; + if (counter == ELEMENT_COUNT) { + // we got the right number of elements + throw new SuccessException(); + } + } + }); + + tryExecute(env, "Read deletes from Kafka"); + + deleteTestTopic(topic); + } // ------------------------------------------------------------------------ // Reading writing test data sets diff --git a/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaITCase.java b/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaITCase.java index 3abbd71876421..07e650ad1d57f 100644 --- a/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaITCase.java +++ b/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaITCase.java @@ -17,7 +17,7 @@ package org.apache.flink.streaming.connectors.kafka; -import org.apache.flink.streaming.util.serialization.DeserializationSchema; +import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema; import org.junit.Test; @@ -28,7 +28,7 @@ public class KafkaITCase extends KafkaConsumerTestBase { @Override - protected FlinkKafkaConsumer getConsumer(List topics, DeserializationSchema deserializationSchema, Properties props) { + protected FlinkKafkaConsumer getConsumer(List topics, KeyedDeserializationSchema deserializationSchema, Properties props) { return new FlinkKafkaConsumer082<>(topics, deserializationSchema, props); } @@ -125,4 +125,9 @@ public void testMultipleTopics() throws Exception { runConsumeMultipleTopics(); } + @Test + public void testAllDeletes() throws Exception { + runAllDeletesTest(); + } + } From 426b92dc82f470b21412e214b37c5230d7ebe963 Mon Sep 17 00:00:00 2001 From: Robert Metzger Date: Mon, 4 Jan 2016 17:00:03 +0100 Subject: [PATCH 2/2] Make the TypeInfoKVSerSchema null aware for the value as well --- .../util/serialization/KeyedDeserializationSchema.java | 2 +- .../TypeInformationKeyValueSerializationSchema.java | 10 +++++++++- 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/KeyedDeserializationSchema.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/KeyedDeserializationSchema.java index 917c2330801be..80bea8de1fe0f 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/KeyedDeserializationSchema.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/KeyedDeserializationSchema.java @@ -35,7 +35,7 @@ public interface KeyedDeserializationSchema extends Serializable, ResultTypeQ * Deserializes the byte message. * * @param messageKey the key as a byte array (null if no key has been set) - * @param message The message, as a byte array. + * @param message The message, as a byte array. (null if the message was empty or deleted) * @param offset the offset of the message in the original source (for example the Kafka offset) * @return The deserialized message as an object. */ diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/TypeInformationKeyValueSerializationSchema.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/TypeInformationKeyValueSerializationSchema.java index ef9cde5bb94c1..250012fa2e479 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/TypeInformationKeyValueSerializationSchema.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/TypeInformationKeyValueSerializationSchema.java @@ -84,7 +84,10 @@ public Tuple2 deserialize(byte[] messageKey, byte[] message, String topic, if(messageKey != null) { key = keySerializer.deserialize(new ByteArrayInputView(messageKey)); } - V value = valueSerializer.deserialize(new ByteArrayInputView(message)); + V value = null; + if(message != null) { + value = valueSerializer.deserialize(new ByteArrayInputView(message)); + } return new Tuple2<>(key, value); } @@ -128,6 +131,11 @@ public byte[] serializeKey(Tuple2 element) { @Override public byte[] serializeValue(Tuple2 element) { + // if the value is null, its serialized value is null as well. + if(element.f1 == null) { + return null; + } + if (valueOutputSerializer == null) { valueOutputSerializer = new DataOutputSerializer(16); }