org.apache.kafka.clients.producer.internals.TransactionalRequestResult#await ( )源码实例Demo

/**
 * Besides committing {@link org.apache.kafka.clients.producer.KafkaProducer#commitTransaction} is also adding new
 * partitions to the transaction. flushNewPartitions method is moving this logic to pre-commit/flush, to make
 * resumeTransaction simpler. Otherwise resumeTransaction would require to restore state of the not yet added/"in-flight"
 * partitions.
 */
private void flushNewPartitions() {
	LOG.info("Flushing new partitions");
	TransactionalRequestResult result = enqueueNewPartitions();
	Object sender = getValue(kafkaProducer, "sender");
	invoke(sender, "wakeup");
	result.await();
}
 

/**
 * Besides committing {@link org.apache.kafka.clients.producer.KafkaProducer#commitTransaction} is also adding new
 * partitions to the transaction. flushNewPartitions method is moving this logic to pre-commit/flush, to make
 * resumeTransaction simpler. Otherwise resumeTransaction would require to restore state of the not yet added/"in-flight"
 * partitions.
 */
private void flushNewPartitions() {
	LOG.info("Flushing new partitions");
	TransactionalRequestResult result = enqueueNewPartitions();
	Object sender = getValue(kafkaProducer, "sender");
	invoke(sender, "wakeup");
	result.await();
}
 

/**
 * Besides committing {@link org.apache.kafka.clients.producer.KafkaProducer#commitTransaction} is also adding new
 * partitions to the transaction. flushNewPartitions method is moving this logic to pre-commit/flush, to make
 * resumeTransaction simpler. Otherwise resumeTransaction would require to restore state of the not yet added/"in-flight"
 * partitions.
 */
private void flushNewPartitions() {
	LOG.info("Flushing new partitions");
	TransactionalRequestResult result = enqueueNewPartitions();
	Object sender = getValue(kafkaProducer, "sender");
	invoke(sender, "wakeup");
	result.await();
}
 

/**
 * Besides committing {@link org.apache.kafka.clients.producer.KafkaProducer#commitTransaction} is also adding new
 * partitions to the transaction. flushNewPartitions method is moving this logic to pre-commit/flush, to make
 * resumeTransaction simpler. Otherwise resumeTransaction would require to restore state of the not yet added/"in-flight"
 * partitions.
 */
private void flushNewPartitions() {
	LOG.info("Flushing new partitions");
	TransactionalRequestResult result = enqueueNewPartitions();
	Object sender = getValue(kafkaProducer, "sender");
	invoke(sender, "wakeup");
	result.await();
}
 

/**
 * Besides committing {@link org.apache.kafka.clients.producer.KafkaProducer#commitTransaction} is also adding new
 * partitions to the transaction. flushNewPartitions method is moving this logic to pre-commit/flush, to make
 * resumeTransaction simpler. Otherwise resumeTransaction would require to restore state of the not yet added/"in-flight"
 * partitions.
 */
private void flushNewPartitions() {
	LOG.info("Flushing new partitions");
	TransactionalRequestResult result = enqueueNewPartitions();
	Object sender = getField(kafkaProducer, "sender");
	invoke(sender, "wakeup");
	result.await();
}
 

/**
 * Besides committing {@link org.apache.kafka.clients.producer.KafkaProducer#commitTransaction} is also adding new
 * partitions to the transaction. flushNewPartitions method is moving this logic to pre-commit/flush, to make
 * resumeTransaction simpler. Otherwise resumeTransaction would require to restore state of the not yet added/"in-flight"
 * partitions.
 */
private void flushNewPartitions() {
	LOG.info("Flushing new partitions");
	TransactionalRequestResult result = enqueueNewPartitions();
	Object sender = getValue(kafkaProducer, "sender");
	invoke(sender, "wakeup");
	result.await();
}