code: public void resumeTransaction(long producerId, short epoch) {
  		Preconditions.checkState(producerId >= 0 && epoch >= 0, "Incorrect values for producerId {} and epoch {}", producerId, epoch);
  		LOG.info("Attempting to resume transaction {} with producerId {} and epoch {}", transactionalId, producerId, epoch);
  
  		Object transactionManager = getValue(kafkaProducer, "transactionManager");
  		synchronized (transactionManager) {
  			Object sequenceNumbers = getValue(transactionManager, "sequenceNumbers");
  
  			invoke(transactionManager, "transitionTo", getEnum("org.apache.kafka.clients.producer.internals.TransactionManager$State.INITIALIZING"));
  			invoke(sequenceNumbers, "clear");
  
  			Object producerIdAndEpoch = getValue(transactionManager, "producerIdAndEpoch");
  			setValue(producerIdAndEpoch, "producerId", producerId);
  			setValue(producerIdAndEpoch, "epoch", epoch);
  
  			invoke(transactionManager, "transitionTo", getEnum("org.apache.kafka.clients.producer.internals.TransactionManager$State.READY"));
  
  			invoke(transactionManager, "transitionTo", getEnum("org.apache.kafka.clients.producer.internals.TransactionManager$State.IN_TRANSACTION"));
  			setValue(transactionManager, "transactionStarted", true);
  		}
  	}
docs: Instead of obtaining producerId and epoch from the transaction coordinator, re-use previously obtained ones,
  so that we can resume transaction after a restart. Implementation of this method is based on
  {@link org.apache.kafka.clients.producer.KafkaProducer#initTransactions}.