@Override
    public void initialize() {
        if (eosEnabled) {
            streamsProducer.initTransaction();
        }
    }
@throws StreamsException fatal error that should cause the thread to die (from producer.initTxn)
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java