[ https://issues.apache.org/jira/browse/KAFKA-4691?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15838744#comment-15838744 ]
Matthias J. Sax commented on KAFKA-4691: ---------------------------------------- I see. Understood now. Seems to be a fair change. But I think, we should do it differently to avoid the producer API change. Because Streams does forward producer config provided in {{StreamsConfig}} we can just omit forwarding interceptor configuration -- thus the {{KafkaProducer}} will not have any interceptors configured in the first place -- and Streams just take the interceptors from {{StreamsConfig}} to call them. But I am not sure, if there are any other things to consider, if Streams does the interceptor call instead of the producer. \cc [~apovzner] > ProducerInterceptor.onSend() is called after key and value are serialized > ------------------------------------------------------------------------- > > Key: KAFKA-4691 > URL: https://issues.apache.org/jira/browse/KAFKA-4691 > Project: Kafka > Issue Type: Bug > Components: clients, streams > Affects Versions: 0.10.1.1 > Reporter: Francesco Lemma > Labels: easyfix > Attachments: 2017-01-24 00_50_55-SDG_CR33_DevStudio - Java EE - > org.apache.kafka.streams.processor.internals.Reco.png > > > According to the JavaDoc > (https://kafka.apache.org/0101/javadoc/org/apache/kafka/clients/producer/ProducerInterceptor.html) > " This is called from KafkaProducer.send(ProducerRecord) and > KafkaProducer.send(ProducerRecord, Callback) methods, before key and value > get serialized and partition is assigned (if partition is not specified in > ProducerRecord)". > Although when using this with Kafka Streams > (StreamsConfig.producerPrefix(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG)) the > key and value contained in the record object are already serialized. > As you can see from the screenshot, the serialization is performed inside > RecordCollectionImpl.send(ProducerRecord<K, V> record, Serializer<K> > keySerializer, Serializer<V> valueSerializer, > StreamPartitioner<K, V> partitioner), effectively > before calling the send method of the producer which will trigger the > interceptor. > This makes it unable to perform any kind of operation involving the key or > value of the message, unless at least performing an additional > deserialization. -- This message was sent by Atlassian JIRA (v6.3.4#6332)