sijie commented on a change in pull request #4534: Support Avro schema for
pulsar kafka client wrapper
URL: https://github.com/apache/pulsar/pull/4534#discussion_r294033441
##########
File path:
pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/kafka/clients/producer/PulsarKafkaProducer.java
##########
@@ -244,13 +246,16 @@ public void close(long timeout, TimeUnit unit) {
try {
// Add the partitions info for the new topic
cluster = cluster.withPartitions(readPartitionsInfo(topic));
- List<org.apache.pulsar.client.api.ProducerInterceptor>
wrappedInterceptors = interceptors.stream()
- .map(interceptor -> new
KafkaProducerInterceptorWrapper(interceptor, keySerializer, valueSerializer,
topic))
- .collect(Collectors.toList());
- return pulsarProducerBuilder.clone()
- .topic(topic)
- .intercept(wrappedInterceptors.toArray(new
org.apache.pulsar.client.api.ProducerInterceptor[wrappedInterceptors.size()]))
- .create();
+ if (keySerializer instanceof PulsarKafkaSchema && valueSerializer
instanceof PulsarKafkaSchema) {
Review comment:
> pulsar's own schema does not need to pass through this class,
I don't understand why Pulsar's own schema doesn't need to pass through this
class.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
