Repository: kafka
Updated Branches:
refs/heads/trunk 90b2a2bf6 -> 6cc29e321
MINOR: Add transactionalId in more log lines in the producer.
Author: Apurva Mehta <ap...@confluent.io>
Reviewers: Jason Gustafson <ja...@confluent.io>
Closes #3343 from apurvam/KAFKA-5449-fix-bad-state-transition-in-transaction-manager
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/6cc29e32
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/6cc29e32
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/6cc29e32
Branch: refs/heads/trunk
Commit: 6cc29e32169e069a75304803feac67591a57b6a5
Parents: 90b2a2b
Author: Apurva Mehta <ap...@confluent.io>
Authored: Thu Jun 15 14:28:38 2017 -0700
Committer: Jason Gustafson <ja...@confluent.io>
Committed: Thu Jun 15 14:31:20 2017 -0700
----------------------------------------------------------------------
.../org/apache/kafka/clients/producer/internals/Sender.java | 5 ++++-
.../kafka/clients/producer/internals/TransactionManager.java | 8 +++++---
2 files changed, 9 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/6cc29e32/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
index c9a1292..e2bff27 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
@@ -378,7 +378,10 @@ public class Sender implements Runnable {
private void maybeAbortBatches(RuntimeException exception) {
if (accumulator.hasUnflushedBatches()) {
- log.error("Aborting producer batches due to fatal error", exception);
+ String logPrefix = "";
+ if (transactionManager != null)
+ logPrefix = transactionManager.logPrefix;
+ log.error("{}Aborting producer batches due to fatal error", logPrefix, exception);
accumulator.abortBatches(exception);
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/6cc29e32/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java
index 49738b2..5c498db 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java
@@ -472,9 +472,11 @@ public class TransactionManager {
}
private synchronized void transitionTo(State target, RuntimeException error) {
- if (!currentState.isTransitionValid(currentState, target))
- throw new KafkaException("Invalid transition attempted from state " + currentState.name() +
- " to state " + target.name());
+ if (!currentState.isTransitionValid(currentState, target)) {
+ String idString = transactionalId == null ? "" : "TransactionalId " + transactionalId + ": ";
+ throw new KafkaException(idString + "Invalid transition attempted from state "
+ + currentState.name() + " to state " + target.name());
+ }
if (target == State.FATAL_ERROR || target == State.ABORTABLE_ERROR) {
if (error == null)