diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java index 36a3314e69832..c9aca10563248 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java @@ -62,7 +62,6 @@ import org.apache.kafka.common.requests.MetadataRequest; import org.apache.kafka.common.requests.MetadataResponse; import org.apache.kafka.common.serialization.Deserializer; -import org.apache.kafka.common.serialization.ExtendedDeserializer; import org.apache.kafka.common.utils.CloseableIterator; import org.apache.kafka.common.utils.LogContext; import org.apache.kafka.common.utils.Time; @@ -89,7 +88,6 @@ import java.util.concurrent.atomic.AtomicReference; import static java.util.Collections.emptyList; -import static org.apache.kafka.common.serialization.ExtendedDeserializer.Wrapper.ensureExtended; /** * This class manage the fetching process with the brokers. @@ -112,8 +110,8 @@ public class Fetcher implements SubscriptionState.Listener, Closeable { private final SubscriptionState subscriptions; private final ConcurrentLinkedQueue completedFetches; private final BufferSupplier decompressionBufferSupplier = BufferSupplier.create(); - private final ExtendedDeserializer keyDeserializer; - private final ExtendedDeserializer valueDeserializer; + private final Deserializer keyDeserializer; + private final Deserializer valueDeserializer; private final IsolationLevel isolationLevel; private final Map sessionHandlers; private final AtomicReference cachedListOffsetsException = new AtomicReference<>(); @@ -150,8 +148,8 @@ public Fetcher(LogContext logContext, this.fetchSize = fetchSize; this.maxPollRecords = maxPollRecords; this.checkCrcs = checkCrcs; - this.keyDeserializer = ensureExtended(keyDeserializer); - this.valueDeserializer = ensureExtended(valueDeserializer); + this.keyDeserializer = keyDeserializer; + this.valueDeserializer = valueDeserializer; this.completedFetches = new ConcurrentLinkedQueue<>(); this.sensors = new FetchManagerMetrics(metrics, metricsRegistry); this.retryBackoffMs = retryBackoffMs; diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java index 2a35b30f259dc..b3f76eee49fde 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java @@ -73,7 +73,6 @@ import org.apache.kafka.common.record.AbstractRecords; import org.apache.kafka.common.record.CompressionType; import org.apache.kafka.common.record.RecordBatch; -import org.apache.kafka.common.serialization.ExtendedSerializer; import org.apache.kafka.common.serialization.Serializer; import org.apache.kafka.common.utils.AppInfoParser; import org.apache.kafka.common.utils.KafkaThread; @@ -81,7 +80,6 @@ import org.apache.kafka.common.utils.Time; import org.slf4j.Logger; -import static org.apache.kafka.common.serialization.ExtendedSerializer.Wrapper.ensureExtended; /** * A Kafka client that publishes records to the Kafka cluster. @@ -250,8 +248,8 @@ public class KafkaProducer implements Producer { private final CompressionType compressionType; private final Sensor errors; private final Time time; - private final ExtendedSerializer keySerializer; - private final ExtendedSerializer valueSerializer; + private final Serializer keySerializer; + private final Serializer valueSerializer; private final ProducerConfig producerConfig; private final long maxBlockTimeMs; private final int requestTimeoutMs; @@ -361,20 +359,20 @@ public KafkaProducer(Properties properties, Serializer keySerializer, Seriali this.partitioner = config.getConfiguredInstance(ProducerConfig.PARTITIONER_CLASS_CONFIG, Partitioner.class); long retryBackoffMs = config.getLong(ProducerConfig.RETRY_BACKOFF_MS_CONFIG); if (keySerializer == null) { - this.keySerializer = ensureExtended(config.getConfiguredInstance(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, - Serializer.class)); + this.keySerializer = config.getConfiguredInstance(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, + Serializer.class); this.keySerializer.configure(config.originals(), true); } else { config.ignore(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG); - this.keySerializer = ensureExtended(keySerializer); + this.keySerializer = keySerializer; } if (valueSerializer == null) { - this.valueSerializer = ensureExtended(config.getConfiguredInstance(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, - Serializer.class)); + this.valueSerializer = config.getConfiguredInstance(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, + Serializer.class); this.valueSerializer.configure(config.originals(), false); } else { config.ignore(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG); - this.valueSerializer = ensureExtended(valueSerializer); + this.valueSerializer = valueSerializer; } // load interceptors and make sure they get clientId diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java index ab036a0992cb1..a38bd04f9064b 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java @@ -28,7 +28,6 @@ import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.ProducerFencedException; import org.apache.kafka.common.record.RecordBatch; -import org.apache.kafka.common.serialization.ExtendedSerializer; import org.apache.kafka.common.serialization.Serializer; import java.util.ArrayDeque; @@ -41,8 +40,6 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; -import static org.apache.kafka.common.serialization.ExtendedSerializer.Wrapper.ensureExtended; - /** * A mock of the producer interface you can use for testing code that uses Kafka. *

@@ -59,8 +56,8 @@ public class MockProducer implements Producer { private final Map offsets; private final List>> consumerGroupOffsets; private Map> uncommittedConsumerGroupOffsets; - private final ExtendedSerializer keySerializer; - private final ExtendedSerializer valueSerializer; + private final Serializer keySerializer; + private final Serializer valueSerializer; private boolean autoComplete; private boolean closed; private boolean transactionInitialized; @@ -93,8 +90,8 @@ public MockProducer(final Cluster cluster, this.cluster = cluster; this.autoComplete = autoComplete; this.partitioner = partitioner; - this.keySerializer = ensureExtended(keySerializer); - this.valueSerializer = ensureExtended(valueSerializer); + this.keySerializer = keySerializer; + this.valueSerializer = valueSerializer; this.offsets = new HashMap<>(); this.sent = new ArrayList<>(); this.uncommittedSends = new ArrayList<>(); diff --git a/clients/src/main/java/org/apache/kafka/common/serialization/Deserializer.java b/clients/src/main/java/org/apache/kafka/common/serialization/Deserializer.java index f9eb398cbee23..bc1a714c35ee1 100644 --- a/clients/src/main/java/org/apache/kafka/common/serialization/Deserializer.java +++ b/clients/src/main/java/org/apache/kafka/common/serialization/Deserializer.java @@ -16,6 +16,8 @@ */ package org.apache.kafka.common.serialization; +import org.apache.kafka.common.header.Headers; + import java.io.Closeable; import java.util.Map; @@ -45,6 +47,17 @@ public interface Deserializer extends Closeable { */ T deserialize(String topic, byte[] data); + /** + * Deserialize a record value from a byte array into a value or object. + * @param topic topic associated with the data + * @param headers headers associated with the record; may be empty. + * @param data serialized bytes; may be null; implementations are recommended to handle null by returning a value or null rather than throwing an exception. + * @return deserialized typed data; may be null + */ + default T deserialize(String topic, Headers headers, byte[] data) { + return deserialize(topic, data); + } + @Override void close(); } diff --git a/clients/src/main/java/org/apache/kafka/common/serialization/ExtendedDeserializer.java b/clients/src/main/java/org/apache/kafka/common/serialization/ExtendedDeserializer.java index 2d5be4a148d31..2f4a012fc8f06 100644 --- a/clients/src/main/java/org/apache/kafka/common/serialization/ExtendedDeserializer.java +++ b/clients/src/main/java/org/apache/kafka/common/serialization/ExtendedDeserializer.java @@ -29,7 +29,9 @@ * * A class that implements this interface is expected to have a constructor with no parameters. * @param + * @deprecated This class has been deprecated and will be removed in a future release. Please use {@link Deserializer} instead. */ +@Deprecated public interface ExtendedDeserializer extends Deserializer { /** diff --git a/clients/src/main/java/org/apache/kafka/common/serialization/ExtendedSerializer.java b/clients/src/main/java/org/apache/kafka/common/serialization/ExtendedSerializer.java index 14fbb47b1dbf3..8c949807a03a5 100644 --- a/clients/src/main/java/org/apache/kafka/common/serialization/ExtendedSerializer.java +++ b/clients/src/main/java/org/apache/kafka/common/serialization/ExtendedSerializer.java @@ -29,7 +29,9 @@ * * A class that implements this interface is expected to have a constructor with no parameters. * @param + * @deprecated This class has been deprecated and will be removed in a future release. Please use {@link Serializer} instead. */ +@Deprecated public interface ExtendedSerializer extends Serializer { /** diff --git a/clients/src/main/java/org/apache/kafka/common/serialization/Serializer.java b/clients/src/main/java/org/apache/kafka/common/serialization/Serializer.java index 96fe86b4bd96e..c5d4760d3812b 100644 --- a/clients/src/main/java/org/apache/kafka/common/serialization/Serializer.java +++ b/clients/src/main/java/org/apache/kafka/common/serialization/Serializer.java @@ -16,6 +16,8 @@ */ package org.apache.kafka.common.serialization; +import org.apache.kafka.common.header.Headers; + import java.io.Closeable; import java.util.Map; @@ -46,6 +48,18 @@ public interface Serializer extends Closeable { */ byte[] serialize(String topic, T data); + /** + * Convert {@code data} into a byte array. + * + * @param topic topic associated with data + * @param headers headers associated with the record + * @param data typed data + * @return serialized bytes + */ + default byte[] serialize(String topic, Headers headers, T data) { + return serialize(topic, data); + } + /** * Close this serializer. * diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java index 22fa0a1bfc0f4..676aafdc57d90 100644 --- a/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java @@ -37,6 +37,7 @@ import org.apache.kafka.common.network.Selectable; import org.apache.kafka.common.serialization.ByteArraySerializer; import org.apache.kafka.common.serialization.ExtendedSerializer; +import org.apache.kafka.common.serialization.Serializer; import org.apache.kafka.common.serialization.StringSerializer; import org.apache.kafka.common.utils.MockTime; import org.apache.kafka.common.utils.Time; @@ -416,15 +417,25 @@ public void testTopicRefreshInMetadata() throws Exception { Assert.assertTrue("Topic should still exist in metadata", metadata.containsTopic(topic)); } + @SuppressWarnings("unchecked") // safe as generic parameters won't vary + @PrepareOnlyThisForTest(Metadata.class) + @Test + public void testHeadersWithExtendedClasses() throws Exception { + doTestHeaders(ExtendedSerializer.class); + } + + @SuppressWarnings("unchecked") @PrepareOnlyThisForTest(Metadata.class) @Test public void testHeaders() throws Exception { + doTestHeaders(Serializer.class); + } + + private > void doTestHeaders(Class serializerClassToMock) throws Exception { Properties props = new Properties(); props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999"); - @SuppressWarnings("unchecked") // it is safe to suppress, since this is a mock class - ExtendedSerializer keySerializer = PowerMock.createNiceMock(ExtendedSerializer.class); - @SuppressWarnings("unchecked") - ExtendedSerializer valueSerializer = PowerMock.createNiceMock(ExtendedSerializer.class); + T keySerializer = PowerMock.createNiceMock(serializerClassToMock); + T valueSerializer = PowerMock.createNiceMock(serializerClassToMock); KafkaProducer producer = new KafkaProducer<>(props, keySerializer, valueSerializer); Metadata metadata = PowerMock.createNiceMock(Metadata.class); diff --git a/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala b/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala index 983b63cdeebfc..2c0c8a4f40ad5 100644 --- a/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala +++ b/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala @@ -66,61 +66,55 @@ class PlaintextConsumerTest extends BaseConsumerTest { } } - @Test - def testHeadersExtendedSerializerDeserializer() { - val numRecords = 1 - val record = new ProducerRecord(tp.topic, tp.partition, null, "key".getBytes, "value".getBytes) - - val extendedSerializer = new ExtendedSerializer[Array[Byte]] { - - var serializer = new ByteArraySerializer() - - override def serialize(topic: String, headers: Headers, data: Array[Byte]): Array[Byte] = { - headers.add("content-type", "application/octet-stream".getBytes) - serializer.serialize(topic, data) - } + trait SerializerImpl { + var serializer = new ByteArraySerializer() - override def configure(configs: util.Map[String, _], isKey: Boolean): Unit = serializer.configure(configs, isKey) - - override def close(): Unit = serializer.close() - - override def serialize(topic: String, data: Array[Byte]): Array[Byte] = { - fail("method should not be invoked") - null - } + def serialize(topic: String, headers: Headers, data: Array[Byte]): Array[Byte] = { + headers.add("content-type", "application/octet-stream".getBytes) + serializer.serialize(topic, data) } + def configure(configs: util.Map[String, _], isKey: Boolean): Unit = serializer.configure(configs, isKey) - val extendedDeserializer = new ExtendedDeserializer[Array[Byte]] { - - var deserializer = new ByteArrayDeserializer() + def close(): Unit = serializer.close() - override def deserialize(topic: String, headers: Headers, data: Array[Byte]): Array[Byte] = { - val header = headers.lastHeader("content-type") - assertEquals("application/octet-stream", if (header == null) null else new String(header.value())) - deserializer.deserialize(topic, data) - } + def serialize(topic: String, data: Array[Byte]): Array[Byte] = { + fail("method should not be invoked") + null + } + } - override def configure(configs: util.Map[String, _], isKey: Boolean): Unit = deserializer.configure(configs, isKey) + trait DeserializerImpl { + var deserializer = new ByteArrayDeserializer() + def deserialize(topic: String, headers: Headers, data: Array[Byte]): Array[Byte] = { + val header = headers.lastHeader("content-type") + assertEquals("application/octet-stream", if (header == null) null else new String(header.value())) + deserializer.deserialize(topic, data) + } - override def close(): Unit = deserializer.close() + def configure(configs: util.Map[String, _], isKey: Boolean): Unit = deserializer.configure(configs, isKey) - override def deserialize(topic: String, data: Array[Byte]): Array[Byte] = { - fail("method should not be invoked") - null - } + def close(): Unit = deserializer.close() + def deserialize(topic: String, data: Array[Byte]): Array[Byte] = { + fail("method should not be invoked") + null } + } + + private def testHeadersSerializeDeserialize(serializer: Serializer[Array[Byte]], deserializer: Deserializer[Array[Byte]]): Unit = { + val numRecords = 1 + val record = new ProducerRecord(tp.topic, tp.partition, null, "key".getBytes, "value".getBytes) val producer = createProducer( keySerializer = new ByteArraySerializer, - valueSerializer = extendedSerializer) + valueSerializer = serializer) producer.send(record) val consumer = createConsumer( keyDeserializer = new ByteArrayDeserializer, - valueDeserializer = extendedDeserializer) + valueDeserializer = deserializer) assertEquals(0, consumer.assignment.size) consumer.assign(List(tp).asJava) assertEquals(1, consumer.assignment.size) @@ -131,6 +125,22 @@ class PlaintextConsumerTest extends BaseConsumerTest { assertEquals(numRecords, records.size) } + @Test + def testHeadersExtendedSerializerDeserializer(): Unit = { + val extendedSerializer = new ExtendedSerializer[Array[Byte]] with SerializerImpl + val extendedDeserializer = new ExtendedDeserializer[Array[Byte]] with DeserializerImpl + + testHeadersSerializeDeserialize(extendedSerializer, extendedDeserializer) + } + + @Test + def testHeadersSerializerDeserializer(): Unit = { + val extendedSerializer = new Serializer[Array[Byte]] with SerializerImpl + val extendedDeserializer = new Deserializer[Array[Byte]] with DeserializerImpl + + testHeadersSerializeDeserialize(extendedSerializer, extendedDeserializer) + } + @Test def testMaxPollRecords() { val maxPollRecords = 2 diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/ChangedDeserializer.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/ChangedDeserializer.java index 29180f2ddb05e..56193d570ecf9 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/ChangedDeserializer.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/ChangedDeserializer.java @@ -18,29 +18,26 @@ import org.apache.kafka.common.header.Headers; import org.apache.kafka.common.serialization.Deserializer; -import org.apache.kafka.common.serialization.ExtendedDeserializer; import java.nio.ByteBuffer; import java.util.Map; -import static org.apache.kafka.common.serialization.ExtendedDeserializer.Wrapper.ensureExtended; - -public class ChangedDeserializer implements ExtendedDeserializer> { +public class ChangedDeserializer implements Deserializer> { private static final int NEWFLAG_SIZE = 1; - private ExtendedDeserializer inner; + private Deserializer inner; public ChangedDeserializer(final Deserializer inner) { - this.inner = ensureExtended(inner); + this.inner = inner; } - public ExtendedDeserializer inner() { + public Deserializer inner() { return inner; } public void setInner(final Deserializer inner) { - this.inner = ensureExtended(inner); + this.inner = inner; } @Override diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/ChangedSerializer.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/ChangedSerializer.java index 83ac8e039edc9..8a76badbc0aeb 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/ChangedSerializer.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/ChangedSerializer.java @@ -17,23 +17,20 @@ package org.apache.kafka.streams.kstream.internals; import org.apache.kafka.common.header.Headers; -import org.apache.kafka.common.serialization.ExtendedSerializer; import org.apache.kafka.common.serialization.Serializer; import org.apache.kafka.streams.errors.StreamsException; import java.nio.ByteBuffer; import java.util.Map; -import static org.apache.kafka.common.serialization.ExtendedSerializer.Wrapper.ensureExtended; - -public class ChangedSerializer implements ExtendedSerializer> { +public class ChangedSerializer implements Serializer> { private static final int NEWFLAG_SIZE = 1; - private ExtendedSerializer inner; + private Serializer inner; public ChangedSerializer(final Serializer inner) { - this.inner = ensureExtended(inner); + this.inner = inner; } public Serializer inner() { @@ -41,7 +38,7 @@ public Serializer inner() { } public void setInner(final Serializer inner) { - this.inner = ensureExtended(inner); + this.inner = inner; } @Override diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/SourceNode.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/SourceNode.java index 4aaba9e3f088a..d7627fe928876 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/SourceNode.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/SourceNode.java @@ -18,22 +18,19 @@ import org.apache.kafka.common.header.Headers; import org.apache.kafka.common.serialization.Deserializer; -import org.apache.kafka.common.serialization.ExtendedDeserializer; import org.apache.kafka.streams.kstream.internals.ChangedDeserializer; import org.apache.kafka.streams.processor.ProcessorContext; import org.apache.kafka.streams.processor.TimestampExtractor; import java.util.List; -import static org.apache.kafka.common.serialization.ExtendedDeserializer.Wrapper.ensureExtended; - public class SourceNode extends ProcessorNode { private final List topics; private ProcessorContext context; - private ExtendedDeserializer keyDeserializer; - private ExtendedDeserializer valDeserializer; + private Deserializer keyDeserializer; + private Deserializer valDeserializer; private final TimestampExtractor timestampExtractor; public SourceNode(final String name, @@ -44,8 +41,8 @@ public SourceNode(final String name, super(name); this.topics = topics; this.timestampExtractor = timestampExtractor; - this.keyDeserializer = ensureExtended(keyDeserializer); - this.valDeserializer = ensureExtended(valDeserializer); + this.keyDeserializer = keyDeserializer; + this.valDeserializer = valDeserializer; } public SourceNode(final String name, @@ -71,9 +68,9 @@ public void init(final InternalProcessorContext context) { // if deserializers are null, get the default ones from the context if (this.keyDeserializer == null) - this.keyDeserializer = ensureExtended((Deserializer) context.keySerde().deserializer()); + this.keyDeserializer = (Deserializer) context.keySerde().deserializer(); if (this.valDeserializer == null) - this.valDeserializer = ensureExtended((Deserializer) context.valueSerde().deserializer()); + this.valDeserializer = (Deserializer) context.valueSerde().deserializer(); // if value deserializers are for {@code Change} values, set the inner deserializer when necessary if (this.valDeserializer instanceof ChangedDeserializer &&