[jira] [Created] (KAFKA-7678) Failed to close producer due to java.lang.NullPointerException

["Jonathan Santilli (JIRA)" <ji...@apache.org>]

Jonathan Santilli created KAFKA-7678:
----------------------------------------

             Summary: Failed to close producer due to java.lang.NullPointerException
                 Key: KAFKA-7678
                 URL: https://issues.apache.org/jira/browse/KAFKA-7678
             Project: Kafka
          Issue Type: Bug
            Reporter: Jonathan Santilli


This occurs when the group is rebalancing in a Kafka Stream application and the process (the Kafka Stream application) receives a SIGTERM to stop it gracefully.

 

 
{noformat}
ERROR org.apache.kafka.streams.processor.internals.StreamTask - task [1_46] Failed to close producer due to the following error:
java.lang.NullPointerException
 at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.close(RecordCollectorImpl.java:252)
 at org.apache.kafka.streams.processor.internals.StreamTask.maybeAbortTransactionAndCloseRecordCollector(StreamTask.java:607)
 at org.apache.kafka.streams.processor.internals.StreamTask.suspend(StreamTask.java:584)
 at org.apache.kafka.streams.processor.internals.StreamTask.close(StreamTask.java:691)
 at org.apache.kafka.streams.processor.internals.AssignedTasks.closeUnclean(AssignedTasks.java:428)
 at org.apache.kafka.streams.processor.internals.AssignedTasks.close(AssignedTasks.java:408)
 at org.apache.kafka.streams.processor.internals.TaskManager.shutdown(TaskManager.java:260)
 at org.apache.kafka.streams.processor.internals.StreamThread.completeShutdown(StreamThread.java:1181)
 at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:758){noformat}
 

 

Although I have checked the code and the method `maybeAbortTransactionAndCloseRecordCollector` in the `StreamTask.java` class is expecting any kind of error to happen since is catching `Throwable`.

 

 

 
{noformat}
try {
 recordCollector.close();
} catch (final Throwable e) {
 log.error("Failed to close producer due to the following error:", e);
} finally {
 producer = null;
}{noformat}
 

Should we consider this a bug?

In my opinion, we could check for the `null` possibility at `RecordCollectorImpl.java` class:
{noformat}
@Override
public void close() {
 log.debug("Closing producer");
 producer.close();
 producer = null;
 checkForException();
}{noformat}
 

Change it for:

 
{noformat}
@Override
public void close() {
 log.debug("Closing producer");
 if ( Objects.nonNull(producer) ) {
    producer.close();
    producer = null;
 }
 checkForException();
}{noformat}
 

How does that sound?

 

Kafka Brokers running 2.0.0

Kafka Stream and client 2.1.0

OpenJDK 8

 



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)