root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 3 Packages Projects Releases Wiki Activity

KAFKA-10169: Error message when transit to Aborting / AbortableError / FatalError (#8880) 

Reviewers: John Roesler <vvcephei@apache.org>
This commit is contained in:
Guozhang Wang 2020-06-16 10:28:36 -07:00 committed by GitHub
parent e5bb535860
commit bea5ceb434
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 4 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
View File

@ -741,6 +741,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
public void abortTransaction() throws ProducerFencedException {
throwIfNoTransactionManager();
throwIfProducerClosed();
log.info("Aborting incomplete transaction");
TransactionalRequestResult result = transactionManager.beginAbort();
sender.wakeup();
result.await(maxBlockTimeMs, TimeUnit.MILLISECONDS);

2
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java
View File

@ -478,10 +478,12 @@ public class TransactionManager {
return;
}
log.info("Transiting to abortable error state due to {}", exception.toString());
transitionTo(State.ABORTABLE_ERROR, exception);
}
synchronized void transitionToFatalError(RuntimeException exception) {
log.info("Transiting to fatal error state due to {}", exception.toString());
transitionTo(State.FATAL_ERROR, exception);
if (pendingResult != null) {

2
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java
View File

@ -251,7 +251,7 @@ public class RecordCollectorImpl implements RecordCollector {
*/
@Override
public void close() {
log.debug("Closing record collector");
log.info("Closing record collector");
if (eosEnabled) {
streamsProducer.abortTransaction();