[hotfix][kafka] Improve logging in FlinkKafkaProducer011

shuttie · Nov 20, 2017 · 34120ef · 34120ef
1 parent 7c63526
commit 34120ef
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 2 deletions.
diff --git a/...0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java b/...0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java
@@ -1022,7 +1022,12 @@ static class KafkaTransactionState {
 
 		@Override
 		public String toString() {
-			return String.format("%s [transactionalId=%s]", this.getClass().getSimpleName(), transactionalId);
+			return String.format(
+				"%s [transactionalId=%s, producerId=%s, epoch=%s]",
+				this.getClass().getSimpleName(),
+				transactionalId,
+				producerId,
+				epoch);
 		}
 
 		@Override

diff --git a/...rc/main/java/org/apache/flink/streaming/connectors/kafka/internal/FlinkKafkaProducer.java b/...rc/main/java/org/apache/flink/streaming/connectors/kafka/internal/FlinkKafkaProducer.java
@@ -188,7 +188,7 @@ public void flush() {
 	 */
 	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 {}", producerId, epoch);
+		LOG.info("Attempting to resume transaction {} with producerId {} and epoch {}", transactionalId, producerId, epoch);
 
 		Object transactionManager = getValue(kafkaProducer, "transactionManager");
 		synchronized (transactionManager) {