Hello Aravind, We have removed StreamsKafkaClient in since 1.1.0 ( https://issues.apache.org/jira/browse/KAFKA-4857), could you consider upgrading to the newer version and see if this issue goes away? Note that new clients can talk to old versioned brokers since 0.10.1.
Guozhang Guozhang On Wed, Jun 20, 2018 at 10:38 PM, D S Aravind Vedantam <dveda...@yodlee.com> wrote: > Hi All, > We have been using Kafka Streams in our application however have > encountered an issue. We are using Processor API and it works fine for some > time but after a while we encounter this exception that happens in all > Streams threads and processing doesn't happen any further after this > exception happens. Just another variant here is that we have only one > Kafka instance with multiple topics and each topic is being consumed by one > application. > > 2018-06-20 11:58:24,035 ERROR [stderr] (StreamThread-2) Exception in > thread "StreamThread-2" java.lang.NullPointerException > 2018-06-20 11:58:24,035 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.clients.NetworkClient.ready(NetworkClient.java:178) > 2018-06-20 11:58:24,035 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.streams.processor.internals.StreamsKafkaClient. > ensureOneNodeIsReady(StreamsKafkaClient.java:195) > 2018-06-20 11:58:24,035 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.streams.processor.internals.StreamsKafkaClient. > getControllerReadyBrokerId(StreamsKafkaClient.java:215) > 2018-06-20 11:58:24,035 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.streams.processor.internals.StreamsKafkaClient. > createTopics(StreamsKafkaClient.java:158) > 2018-06-20 11:58:24,036 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.streams.processor.internals.InternalTopicManager. > makeReady(InternalTopicManager.java:64) > 2018-06-20 11:58:24,036 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.streams.processor.internals.StreamPartitionAssignor. > prepareTopic(StreamPartitionAssignor.java:618) > 2018-06-20 11:58:24,036 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.streams.processor.internals.StreamPartitionAssignor. > assign(StreamPartitionAssignor.java:449) > 2018-06-20 11:58:24,036 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.clients.consumer.internals.ConsumerCoordinator. > performAssignment(ConsumerCoordinator.java:339) > 2018-06-20 11:58:24,036 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.clients.consumer.internals.AbstractCoordinator. > onJoinLeader(AbstractCoordinator.java:488) > 2018-06-20 11:58:24,036 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.clients.consumer.internals.AbstractCoordinator.access$ > 1100(AbstractCoordinator.java:89) > 2018-06-20 11:58:24,036 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.clients.consumer.internals.AbstractCoordinator$ > JoinGroupResponseHandler.handle(AbstractCoordinator.java:438) > 2018-06-20 11:58:24,036 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.clients.consumer.internals.AbstractCoordinator$ > JoinGroupResponseHandler.handle(AbstractCoordinator.java:420) > 2018-06-20 11:58:24,036 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.clients.consumer.internals.AbstractCoordinator$ > CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:764) > 2018-06-20 11:58:24,036 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.clients.consumer.internals.AbstractCoordinator$ > CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:745) > 2018-06-20 11:58:24,036 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.clients.consumer.internals.RequestFuture$1.onSuccess( > RequestFuture.java:186) > 2018-06-20 11:58:24,036 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.clients.consumer.internals.RequestFuture.fireSuccess( > RequestFuture.java:149) > 2018-06-20 11:58:24,036 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.clients.consumer.internals.RequestFuture.complete( > RequestFuture.java:116) > 2018-06-20 11:58:24,036 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$ > RequestFutureCompletionHandler.fireCompletion(ConsumerNetworkClient.java: > 493) > 2018-06-20 11:58:24,036 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient. > firePendingCompletedRequests(ConsumerNetworkClient.java:322) > 2018-06-20 11:58:24,036 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll( > ConsumerNetworkClient.java:253) > 2018-06-20 11:58:24,036 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll( > ConsumerNetworkClient.java:172) > 2018-06-20 11:58:24,036 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.clients.consumer.internals.AbstractCoordinator. > joinGroupIfNeeded(AbstractCoordinator.java:334) > 2018-06-20 11:58:24,037 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.clients.consumer.internals.AbstractCoordinator. > ensureActiveGroup(AbstractCoordinator.java:303) > 2018-06-20 11:58:24,037 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll( > ConsumerCoordinator.java:286) > 2018-06-20 11:58:24,037 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.clients.consumer.KafkaConsumer. > pollOnce(KafkaConsumer.java:1030) > 2018-06-20 11:58:24,037 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.clients.consumer.KafkaConsumer.poll( > KafkaConsumer.java:995) > 2018-06-20 11:58:24,037 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.streams.processor.internals.StreamThread.runLoop( > StreamThread.java:582) > 2018-06-20 11:58:24,037 ERROR [stderr] (StreamThread-2) at > org.apache.kafka.streams.processor.internals. > StreamThread.run(StreamThread.java:368) > 2018-06-20 12:20:06,292 ERROR [stderr] (StreamThread-1) Exception in > thread "StreamThread-1" java.lang.NullPointerException > 2018-06-20 12:20:06,293 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.clients.NetworkClient.ready(NetworkClient.java:178) > 2018-06-20 12:20:06,293 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.streams.processor.internals.StreamsKafkaClient. > ensureOneNodeIsReady(StreamsKafkaClient.java:195) > 2018-06-20 12:20:06,293 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.streams.processor.internals.StreamsKafkaClient. > getControllerReadyBrokerId(StreamsKafkaClient.java:215) > 2018-06-20 12:20:06,293 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.streams.processor.internals.StreamsKafkaClient. > createTopics(StreamsKafkaClient.java:158) > 2018-06-20 12:20:06,293 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.streams.processor.internals.InternalTopicManager. > makeReady(InternalTopicManager.java:64) > 2018-06-20 12:20:06,293 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.streams.processor.internals.StreamPartitionAssignor. > prepareTopic(StreamPartitionAssignor.java:618) > 2018-06-20 12:20:06,293 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.streams.processor.internals.StreamPartitionAssignor. > assign(StreamPartitionAssignor.java:449) > 2018-06-20 12:20:06,293 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.clients.consumer.internals.ConsumerCoordinator. > performAssignment(ConsumerCoordinator.java:339) > 2018-06-20 12:20:06,293 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.clients.consumer.internals.AbstractCoordinator. > onJoinLeader(AbstractCoordinator.java:488) > 2018-06-20 12:20:06,293 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.clients.consumer.internals.AbstractCoordinator.access$ > 1100(AbstractCoordinator.java:89) > 2018-06-20 12:20:06,293 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.clients.consumer.internals.AbstractCoordinator$ > JoinGroupResponseHandler.handle(AbstractCoordinator.java:438) > 2018-06-20 12:20:06,293 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.clients.consumer.internals.AbstractCoordinator$ > JoinGroupResponseHandler.handle(AbstractCoordinator.java:420) > 2018-06-20 12:20:06,293 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.clients.consumer.internals.AbstractCoordinator$ > CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:764) > 2018-06-20 12:20:06,293 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.clients.consumer.internals.AbstractCoordinator$ > CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:745) > 2018-06-20 12:20:06,293 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.clients.consumer.internals.RequestFuture$1.onSuccess( > RequestFuture.java:186) > 2018-06-20 12:20:06,293 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.clients.consumer.internals.RequestFuture.fireSuccess( > RequestFuture.java:149) > 2018-06-20 12:20:06,293 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.clients.consumer.internals.RequestFuture.complete( > RequestFuture.java:116) > 2018-06-20 12:20:06,293 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$ > RequestFutureCompletionHandler.fireCompletion(ConsumerNetworkClient.java: > 493) > 2018-06-20 12:20:06,294 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient. > firePendingCompletedRequests(ConsumerNetworkClient.java:322) > 2018-06-20 12:20:06,294 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll( > ConsumerNetworkClient.java:253) > 2018-06-20 12:20:06,294 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll( > ConsumerNetworkClient.java:172) > 2018-06-20 12:20:06,294 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.clients.consumer.internals.AbstractCoordinator. > joinGroupIfNeeded(AbstractCoordinator.java:334) > 2018-06-20 12:20:06,294 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.clients.consumer.internals.AbstractCoordinator. > ensureActiveGroup(AbstractCoordinator.java:303) > 2018-06-20 12:20:06,294 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll( > ConsumerCoordinator.java:286) > 2018-06-20 12:20:06,294 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.clients.consumer.KafkaConsumer. > pollOnce(KafkaConsumer.java:1030) > 2018-06-20 12:20:06,294 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.clients.consumer.KafkaConsumer.poll( > KafkaConsumer.java:995) > 2018-06-20 12:20:06,294 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.streams.processor.internals.StreamThread.runLoop( > StreamThread.java:582) > 2018-06-20 12:20:06,294 ERROR [stderr] (StreamThread-1) at > org.apache.kafka.streams.processor.internals. > StreamThread.run(StreamThread.java:368) > > Any inputs on this would very helpful. > > Regards, > Aravind. > -- -- Guozhang
