Uploaded image for project: 'Nuxeo Platform'
  1. Nuxeo Platform
  2. NXP-30089

Possible NPE in KafkaLogTailer.close

    XMLWordPrintable

    Details

    • Type: Bug
    • Status: Resolved
    • Priority: Minor
    • Resolution: Fixed
    • Affects Version/s: 11.1
    • Fix Version/s: 11.5, 2021.2
    • Component/s: Streams, Tests
    • Tags:
    • Team:
      PLATFORM
    • Sprint:
      nxplatform #27
    • Story Points:
      1

      Description

      During unit test that kills thread pool the following NPE has been observed:

      Regression

Run unit tests / Run postgresql unit tests / org.nuxeo.lib.stream.tests.pattern.TestPatternQueuingKafka.killConsumers
Failing for the past 1 build (Since Failed#66 )
Took 0.37 sec.
add description
Stacktrace

java.lang.NullPointerException
	at org.nuxeo.lib.stream.log.kafka.KafkaLogTailer.close(KafkaLogTailer.java:446)
	at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
	at java.base/java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:177)
	at java.base/java.util.concurrent.ConcurrentHashMap$KeySpliterator.forEachRemaining(ConcurrentHashMap.java:3566)
	at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:484)
	at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:474)
	at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150)
	at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173)
	at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
	at java.base/java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:497)
	at org.nuxeo.lib.stream.log.internals.AbstractLogManager.close(AbstractLogManager.java:221)
	at org.nuxeo.lib.stream.log.kafka.KafkaLogManager.close(KafkaLogManager.java:148)
	at org.nuxeo.lib.stream.tests.pattern.TestPatternQueuing.closeManager(TestPatternQueuing.java:68)
	at org.nuxeo.lib.stream.tests.pattern.TestPatternQueuing.resetManager(TestPatternQueuing.java:74)
	at org.nuxeo.lib.stream.tests.pattern.TestPatternQueuing.killConsumers(TestPatternQueuing.java:168)

      output 

      2021-01-20 15:47:35,918 [main] INFO  [KafkaUtils] Creating topic: nuxeo-test-1611157655910-killConsumers, partitions: 2, replications: 1
2021-01-20 15:47:35,981 [main] INFO  [ConsumerPool] Creating consumer pool using Log subscribe on killConsumers
2021-01-20 15:47:35,982 [Nuxeo-ConsumerPool-00] WARN  [AbstractCallablePool] Start Nuxeo-Consumer Pool on 2 thread(s).
2021-01-20 15:47:35,997 [Nuxeo-ConsumerPool-00] INFO  [AbstractCallablePool] Pool is up and running
2021-01-20 15:47:36,204 [Nuxeo-Consumer-00] INFO  [KafkaLogTailer] Rebalance assigned: [killConsumers-00]
2021-01-20 15:47:36,204 [Nuxeo-Consumer-01] INFO  [KafkaLogTailer] Rebalance assigned: [killConsumers-01]
2021-01-20 15:47:36,212 [Nuxeo-Consumer-00-0-rebalance-assigned-KafkaLogTailer{id=default:killConsumers-00, closed=false, codec=org.nuxeo.lib.stream.codec.NoCodec@c3c4c1c}] WARN  [ConsumerRunner] Rollback current batch because of consumer rebalancing
2021-01-20 15:47:36,212 [Nuxeo-Consumer-00-0-rebalance-assigned-KafkaLogTailer{id=default:killConsumers-00, closed=false, codec=org.nuxeo.lib.stream.codec.NoCodec@c3c4c1c}] INFO  [ConsumerRunner] Rebalance
2021-01-20 15:47:36,248 [Nuxeo-ConsumerPool-00] ERROR [AbstractCallablePool] End of consumer interrupted.
2021-01-20 15:47:36,248 [Nuxeo-Consumer-00-3-foo] WARN  [ConsumerRunner] Rollback batch
java.lang.InterruptedException: java.lang.InterruptedException
	at org.nuxeo.lib.stream.log.kafka.KafkaLogTailer.poll(KafkaLogTailer.java:220) ~[nuxeo-stream-2021.1.20.jar:?]
	at org.nuxeo.lib.stream.log.kafka.KafkaLogTailer.read(KafkaLogTailer.java:175) ~[nuxeo-stream-2021.1.20.jar:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.ConsumerRunner.acceptBatch(ConsumerRunner.java:304) ~[classes/:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.ConsumerRunner.processBatch(ConsumerRunner.java:256) ~[classes/:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.ConsumerRunner.processBatchWithRetry(ConsumerRunner.java:218) ~[classes/:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.ConsumerRunner.consumerLoop(ConsumerRunner.java:202) ~[classes/:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.ConsumerRunner.call(ConsumerRunner.java:147) ~[classes/:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.ConsumerRunner.call(ConsumerRunner.java:60) ~[classes/:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.AbstractCallablePool.lambda$runPool$1(AbstractCallablePool.java:99) ~[classes/:?]
	at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1700) [?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
	at java.lang.Thread.run(Thread.java:834) [?:?]
2021-01-20 15:47:36,248 [Nuxeo-Consumer-01-0-rebalance-assigned-KafkaLogTailer{id=default:killConsumers-01, closed=false, codec=org.nuxeo.lib.stream.codec.NoCodec@c3c4c1c}] WARN  [ConsumerRunner] Rollback batch
java.lang.InterruptedException: java.lang.InterruptedException
	at org.nuxeo.lib.stream.log.kafka.KafkaLogTailer.poll(KafkaLogTailer.java:220) ~[nuxeo-stream-2021.1.20.jar:?]
	at org.nuxeo.lib.stream.log.kafka.KafkaLogTailer.read(KafkaLogTailer.java:175) ~[nuxeo-stream-2021.1.20.jar:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.ConsumerRunner.acceptBatch(ConsumerRunner.java:304) ~[classes/:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.ConsumerRunner.processBatch(ConsumerRunner.java:256) ~[classes/:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.ConsumerRunner.processBatchWithRetry(ConsumerRunner.java:218) ~[classes/:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.ConsumerRunner.consumerLoop(ConsumerRunner.java:202) ~[classes/:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.ConsumerRunner.call(ConsumerRunner.java:147) ~[classes/:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.ConsumerRunner.call(ConsumerRunner.java:60) ~[classes/:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.AbstractCallablePool.lambda$runPool$1(AbstractCallablePool.java:99) ~[classes/:?]
	at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1700) [?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
	at java.lang.Thread.run(Thread.java:834) [?:?]
2021-01-20 15:47:36,249 [Nuxeo-Consumer-00-3-foo] WARN  [AbstractCoordinator] [Consumer clientId=nuxeo-test-1611157655910-default-29, groupId=nuxeo-test-1611157655910-default] Interrupted while waiting for consumer heartbeat thread to close
2021-01-20 15:47:36,249 [Nuxeo-Consumer-01-0-rebalance-assigned-KafkaLogTailer{id=default:killConsumers-01, closed=false, codec=org.nuxeo.lib.stream.codec.NoCodec@c3c4c1c}] WARN  [AbstractCoordinator] [Consumer clientId=nuxeo-test-1611157655910-default-30, groupId=nuxeo-test-1611157655910-default] Interrupted while waiting for consumer heartbeat thread to close
2021-01-20 15:47:36,249 [Nuxeo-Consumer-00-3-foo] INFO  [KafkaLogTailer] Rebalance revoked: [killConsumers-00]
2021-01-20 15:47:36,250 [Nuxeo-ConsumerPool-00] ERROR [AbstractCallablePool] End of consumer interrupted.
2021-01-20 15:47:36,250 [Nuxeo-Consumer-01-0-rebalance-assigned-KafkaLogTailer{id=default:killConsumers-01, closed=false, codec=org.nuxeo.lib.stream.codec.NoCodec@c3c4c1c}] INFO  [KafkaLogTailer] Rebalance revoked: [killConsumers-01]
2021-01-20 15:47:36,250 [Nuxeo-ConsumerPool-00] INFO  [ArrayList] Consumer status FAILURE
2021-01-20 15:47:36,250 [Nuxeo-ConsumerPool-00] INFO  [ArrayList] Consumer status FAILURE
2021-01-20 15:47:36,250 [Nuxeo-ConsumerPool-00] WARN  [ConsumerPool] Consumers status: threads: 2, failure 2, messages committed: 0, elapsed: 0.00s, throughput: 0.00 msg/s
2021-01-20 15:47:36,250 [Nuxeo-Consumer-00-4-rebalance-revoked] ERROR [KafkaConsumer] [Consumer clientId=nuxeo-test-1611157655910-default-29, groupId=nuxeo-test-1611157655910-default] Failed to close coordinator
org.apache.kafka.common.errors.InterruptException: java.lang.InterruptedException
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeThrowInterruptException(ConsumerNetworkClient.java:520) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:281) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.pollNoWakeup(ConsumerNetworkClient.java:306) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.maybeLeaveGroup(AbstractCoordinator.java:1013) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.close(AbstractCoordinator.java:977) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.close(ConsumerCoordinator.java:895) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2373) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2340) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2290) ~[kafka-clients-2.6.0.jar:?]
	at org.nuxeo.lib.stream.log.kafka.KafkaLogTailer.close(KafkaLogTailer.java:442) ~[nuxeo-stream-2021.1.20.jar:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.ConsumerRunner.call(ConsumerRunner.java:151) ~[classes/:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.ConsumerRunner.call(ConsumerRunner.java:60) ~[classes/:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.AbstractCallablePool.lambda$runPool$1(AbstractCallablePool.java:99) ~[classes/:?]
	at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1700) [?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
	at java.lang.Thread.run(Thread.java:834) [?:?]
Caused by: java.lang.InterruptedException
	... 17 more
2021-01-20 15:47:36,250 [Nuxeo-Consumer-01-0-rebalance-revoked] ERROR [KafkaConsumer] [Consumer clientId=nuxeo-test-1611157655910-default-30, groupId=nuxeo-test-1611157655910-default] Failed to close coordinator
org.apache.kafka.common.errors.InterruptException: java.lang.InterruptedException
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeThrowInterruptException(ConsumerNetworkClient.java:520) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:281) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.pollNoWakeup(ConsumerNetworkClient.java:306) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.maybeLeaveGroup(AbstractCoordinator.java:1013) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.close(AbstractCoordinator.java:977) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.close(ConsumerCoordinator.java:895) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2373) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2340) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2290) ~[kafka-clients-2.6.0.jar:?]
	at org.nuxeo.lib.stream.log.kafka.KafkaLogTailer.close(KafkaLogTailer.java:442) ~[nuxeo-stream-2021.1.20.jar:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.ConsumerRunner.call(ConsumerRunner.java:151) ~[classes/:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.ConsumerRunner.call(ConsumerRunner.java:60) ~[classes/:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.AbstractCallablePool.lambda$runPool$1(AbstractCallablePool.java:99) ~[classes/:?]
	at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1700) [?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
	at java.lang.Thread.run(Thread.java:834) [?:?]
Caused by: java.lang.InterruptedException
	... 17 more
2021-01-20 15:47:36,254 [Nuxeo-Consumer-00-4-rebalance-revoked] INFO  [KafkaLogTailer] Discard error while closing consumer: 
org.apache.kafka.common.errors.InterruptException: java.lang.InterruptedException
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeThrowInterruptException(ConsumerNetworkClient.java:520) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:281) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.pollNoWakeup(ConsumerNetworkClient.java:306) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.maybeLeaveGroup(AbstractCoordinator.java:1013) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.close(AbstractCoordinator.java:977) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.close(ConsumerCoordinator.java:895) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2373) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2340) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2290) ~[kafka-clients-2.6.0.jar:?]
	at org.nuxeo.lib.stream.log.kafka.KafkaLogTailer.close(KafkaLogTailer.java:442) ~[nuxeo-stream-2021.1.20.jar:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.ConsumerRunner.call(ConsumerRunner.java:151) ~[classes/:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.ConsumerRunner.call(ConsumerRunner.java:60) ~[classes/:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.AbstractCallablePool.lambda$runPool$1(AbstractCallablePool.java:99) ~[classes/:?]
	at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1700) [?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
	at java.lang.Thread.run(Thread.java:834) [?:?]
Caused by: java.lang.InterruptedException
	... 17 more
2021-01-20 15:47:36,254 [Nuxeo-Consumer-01-0-rebalance-revoked] INFO  [KafkaLogTailer] Discard error while closing consumer: 
org.apache.kafka.common.errors.InterruptException: java.lang.InterruptedException
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeThrowInterruptException(ConsumerNetworkClient.java:520) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:281) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.pollNoWakeup(ConsumerNetworkClient.java:306) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.maybeLeaveGroup(AbstractCoordinator.java:1013) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.close(AbstractCoordinator.java:977) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.close(ConsumerCoordinator.java:895) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2373) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2340) ~[kafka-clients-2.6.0.jar:?]
	at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2290) ~[kafka-clients-2.6.0.jar:?]
	at org.nuxeo.lib.stream.log.kafka.KafkaLogTailer.close(KafkaLogTailer.java:442) ~[nuxeo-stream-2021.1.20.jar:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.ConsumerRunner.call(ConsumerRunner.java:151) ~[classes/:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.ConsumerRunner.call(ConsumerRunner.java:60) ~[classes/:?]
	at org.nuxeo.lib.stream.pattern.consumer.internals.AbstractCallablePool.lambda$runPool$1(AbstractCallablePool.java:99) ~[classes/:?]
	at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1700) [?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
	at java.lang.Thread.run(Thread.java:834) [?:?]
Caused by: java.lang.InterruptedException
	... 17 more
2021-01-20 15:47:36,255 [main] INFO  [KafkaLogTailer] Closing tailer from another thread, send wakeup
2021-01-20 15:47:36,257 [main] INFO  [KafkaUtils] Deleting topic: nuxeo-test-1611157655910-killConsumers

      Seen here:
      https://jenkins.platform.dev.nuxeo.com/job/nuxeo/job/LTS/job/nuxeo/job/2021/66/

        Attachments

          Issue Links

          is related to

          Bug - A problem which impairs or prevents the functions of the product. NXP-30967 Possible NPE in KafkaLogTailer.close (bis)

          • Minor - Minor loss of function, or other problem where easy workaround is present.
          • Resolved

            Activity

              People

              • Votes:
                0 Vote for this issue
                Watchers:
                2 Start watching this issue

                Dates

                • Created:
                  Updated:
                  Resolved:

                  Time Tracking

                  Estimated:
                  Original Estimate - Not Specified
                  Not Specified
                  Remaining:
                  Remaining Estimate - 0 minutes
                  0m
                  Logged:
                  Time Spent - 2 hours
                  2h