[KAFKA-3142] Improve error message in kstreams

• Type: Sub-task
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 0.10.0.0
• Assignee: Guozhang Wang
• Reporter: Jay Kreps
• Created: 23/Jan/16 22:56
• Updated: 27/Jan/16 07:09
• Resolved: 27/Jan/16 07:09

Description

If you have a bug in your kstream code the error message could be slightly better.

1. Not sure what the second error about it already being closed is.
2. I recommend we avoid wrapping the exception (at least for runtime exceptions) so it is more clear to the user that the error is coming from their code not ours.

[2016-01-23 14:40:59,776] ERROR Uncaught error during processing in thread [StreamThread-1]:  (org.apache.kafka.streams.processor.internals.StreamThread:222)
org.apache.kafka.common.KafkaException: java.lang.NullPointerException
	at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:327)
	at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:220)
Caused by: java.lang.NullPointerException
	at TestKstreamsApi.lambda$0(TestKstreamsApi.java:27)
	at org.apache.kafka.streams.kstream.internals.KStreamMap$KStreamMapProcessor.process(KStreamMap.java:42)
	at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:68)
	at org.apache.kafka.streams.processor.internals.StreamTask.forward(StreamTask.java:324)
	at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:175)
	at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:56)
	at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:170)
	at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:310)
	... 1 more
[2016-01-23 14:40:59,803] ERROR Failed to close a StreamTask #0_0 in thread [StreamThread-1]:  (org.apache.kafka.streams.processor.internals.StreamThread:586)
java.lang.IllegalStateException: This consumer has already been closed.
	at org.apache.kafka.clients.consumer.KafkaConsumer.ensureNotClosed(KafkaConsumer.java:1281)
	at org.apache.kafka.clients.consumer.KafkaConsumer.acquire(KafkaConsumer.java:1292)
	at org.apache.kafka.clients.consumer.KafkaConsumer.assign(KafkaConsumer.java:779)
	at org.apache.kafka.streams.processor.internals.ProcessorStateManager.close(ProcessorStateManager.java:339)
	at org.apache.kafka.streams.processor.internals.AbstractTask.close(AbstractTask.java:95)
	at org.apache.kafka.streams.processor.internals.StreamTask.close(StreamTask.java:306)
	at org.apache.kafka.streams.processor.internals.StreamThread.closeOne(StreamThread.java:584)
	at org.apache.kafka.streams.processor.internals.StreamThread.removeStreamTasks(StreamThread.java:571)
	at org.apache.kafka.streams.processor.internals.StreamThread.shutdown(StreamThread.java:265)
	at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:225)