From a2a3c46e0c109614c88e3caa4d2114a844a1a75b Mon Sep 17 00:00:00 2001
From: Aviem Zur <aviemzur@gmail.com>
Date: Fri, 17 Feb 2017 12:55:43 +0200
Subject: [PATCH] More informative KafkaCheckpointMark toString

---
 .../beam/sdk/io/kafka/KafkaCheckpointMark.java   | 16 ++++++++++++++++
 1 file changed, 16 insertions(+)

diff --git a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaCheckpointMark.java b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaCheckpointMark.java
index 763a98a9b667..61a382d6ee2a 100644
--- a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaCheckpointMark.java
+++ b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaCheckpointMark.java
@@ -17,6 +17,8 @@
  */
 package org.apache.beam.sdk.io.kafka;
 
+import com.google.common.base.Joiner;
+
 import java.io.IOException;
 import java.io.Serializable;
 import java.util.List;
@@ -52,6 +54,11 @@ public void finalizeCheckpoint() throws IOException {
     // is restarted (checkpoint is not available for job restarts).
   }
 
+  @Override
+  public String toString() {
+    return "KafkaCheckpointMark{partitions=" + Joiner.on(",").join(partitions) + '}';
+  }
+
   /**
    * A tuple to hold topic, partition, and offset that comprise the checkpoint
    * for a single partition.
@@ -80,6 +87,15 @@ public int getPartition() {
     public long getNextOffset() {
       return nextOffset;
     }
+
+    @Override
+    public String toString() {
+      return "PartitionMark{"
+          + "topic='" + topic + '\''
+          + ", partition=" + partition
+          + ", nextOffset=" + nextOffset
+          + '}';
+    }
   }
 }