flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/FlinkKafkaProducer.java
/**
	 * 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}.
	 */
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
org.apache.kafka.clients.producer.KafkaProducer#initTransactions.
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 org . apache . kafka . clients . producer . KafkaProducer#initTransactions .
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);
		}
	}
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 ) ; } }
Instead of obtaining producerId and epoch from the transaction coordinator, re-use previously obtained ones,
 so that we can resume transaction after a restart.
Instead of obtaining producerId and epoch from the transaction coordinator re - use previously obtained ones so that we can resume transaction after a restart .