Allow parallel consumer to work with subclasses of Apache KafkaConsumer

[mrabhishek]

I do extend the stock Apache KafkaConsumer class to add some additional behavior to its existing poll method. However, since the parallel-consumer expects to read private fields from the KafkaConsumer class, it does not work with an instance of my subclass provided as a consumer.

The error is specifically at this line.

if (consumer instanceof KafkaConsumer) {
     // Could use Commons Lang FieldUtils#readField - but, avoid needing commons lang
     Field coordinatorField = KafkaConsumer.class.getDeclaredField("coordinator");

If we modified this check to add looking into current and its parent class(es), things will work fine. Let me know if that is an acceptable fix and I'll push a PR.

For details - My Consumer looks something like this.

public class CustomKafkaConsumer<K, V> extends KafkaConsumer<K, V> {
}

and I want to use this with parallel consumer like this.

ParallelConsumerOptions.<String, byte[]>builder()
                .consumer(new CustomKafkaConsumer())
                .maxConcurrency(2)
                .ordering(ParallelConsumerOptions.ProcessingOrder.UNORDERED)
                .commitMode(ParallelConsumerOptions.CommitMode.PERIODIC_CONSUMER_SYNC)
                .build();

[mrabhishek]

This seems to have been fixed in 0.5.2.5 as part of #195 . Closing as duplicate.

[Kirikx]

@mrabhishek No, the problem has not been resolved. I get a similar problem when using spring boot kafka (PC version: 0.5.2.5):
c.p.i.AbstractParallelEoSStreamProcessor : Consumer is neither a KafkaConsumer nor a MockConsumer - cannot check auto commit is disabled for consumer type: brave.kafka.clients.TracingConsumer