title: Why did the Kafka stream fail to produce data after a long time?
body: <p>KafkaStream failed to produce data after a long time. (Exceeding the set expiration time)</p>
  
  <p>Even KafkaStream was dead after logging error message.</p>
  
  <p>Exception is below:</p>
  
  <pre><code>org.apache.kafka.common.KafkaException: Cannot execute transactional method because we are in an error state
  at org.apache.kafka.clients.producer.internals.TransactionManager.maybeFailWithError(TransactionManager.java:784)
  at org.apache.kafka.clients.producer.internals.TransactionManager.beginAbort(TransactionManager.java:229)
  at org.apache.kafka.clients.producer.KafkaProducer.abortTransaction(KafkaProducer.java:660)
  at org.apache.kafka.streams.processor.internals.StreamTask.closeSuspended(StreamTask.java:493)
  at org.apache.kafka.streams.processor.internals.StreamTask.close(StreamTask.java:553)
  at org.apache.kafka.streams.processor.internals.AssignedTasks.close(AssignedTasks.java:405)
  at org.apache.kafka.streams.processor.internals.TaskManager.shutdown(TaskManager.java:260)
  at org.apache.kafka.streams.processor.internals.StreamThread.completeShutdown(StreamThread.java:1111)
  at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:730)
  org.apache.kafka.common.KafkaException: Unexpected error in AddOffsetsToTxnResponse: The producer attempted to use a producer id which is not currently assigned to its transactional id
  at org.apache.kafka.clients.producer.internals.TransactionManager$AddOffsetsToTxnHandler.handleResponse(TransactionManager.java:1237)
  at org.apache.kafka.clients.producer.internals.TransactionManager$TxnRequestHandler.onComplete(TransactionManager.java:907)
  at org.apache.kafka.clients.ClientResponse.onComplete(ClientResponse.java:101)
  at org.apache.kafka.clients.NetworkClient.completeResponses(NetworkClient.java:482)
  at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:474)
  at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:216)
  at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:163)
  at java.lang.Thread.run(Thread.java:834)
  </code></pre>
  
  <p>Version:</p>
  
  <ol>
  <li>Kafka Broker: 2.0.0</li>
  <li>kafka-clients: 1.1.1</li>
  <li>kafka-streams: 1.1.1</li>
  </ol>
  
  <p>(Broker and Producer) Options are all default:</p>
  
  <ol>
  <li>TRANSACTION_TIMEOUT_CONFIG</li>
  <li>transactional.id.expiration.ms</li>
  <li>transaction.max.timeout.ms</li>
  </ol>
  
  <p>Code:</p>
  
  <pre><code>Properties properties = new Properties();
  properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
  properties.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE);
  
  StreamsBuilder builder = new StreamsBuilder();
  builder.stream("from", Consumed.with(Serdes.Integer(), Serdes.String()))
         .peek((key, value) -&gt; System.out.println(value))
         .to("to", Produced.with(Serdes.Integer(), Serdes.String()), (key, value, numPartitions) -&gt; key % numPartitions));
  
  KafkaStreams streams = new KafkaStreams(bulider.build(), properties);
  stream.start();
  </code></pre>