Kafka Streams - Stopped on stopped broker

stderr

Exception in thread "StreamThread-1" org.apache.kafka.streams.errors.StreamsException: Exception caught in process. taskId=0_23, processor=KSTREAM-SOURCE-0000000000, topic=abc, partition=23, offset=388592
	at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:216)
	at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:641)
	at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:368)
Caused by: org.apache.kafka.streams.errors.StreamsException: task [0_23] exception caught when producing
	at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.checkForException(RecordCollectorImpl.java:119)
	at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:76)
	at org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:79)
	at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:83)
	at org.apache.kafka.streams.kstream.internals.KStreamFlatMap$KStreamFlatMapProcessor.process(KStreamFlatMap.java:43)
	at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:48)
	at org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:188)
	at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:134)
	at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:83)
	at org.apache.kafka.streams.kstream.internals.KStreamPassThrough$KStreamPassThroughProcessor.process(KStreamPassThrough.java:34)
	at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:48)
	at org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:188)
	at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:134)
	at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:97)
	at org.apache.kafka.streams.kstream.internals.KStreamBranch$KStreamBranchProcessor.process(KStreamBranch.java:46)
	at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:48)
	at org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:188)
	at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:134)
	at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:83)
	at org.apache.kafka.streams.kstream.internals.KStreamTransform$KStreamTransformProcessor.process(KStreamTransform.java:60)
	at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:48)
	at org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:188)
	at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:134)
	at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:83)
	at org.apache.kafka.streams.kstream.internals.KStreamMap$KStreamMapProcessor.process(KStreamMap.java:43)
	at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:48)
	at org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:188)
	at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:134)
	at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:83)
	at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:70)
	at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:197)
	... 2 more
Caused by: org.apache.kafka.common.errors.NotLeaderForPartitionException: This server is not the leader for that topic-partition.