The OutboxProducer restores the sequence-number from the wirtten topic

[demos/kafka/outbox] / delivery / src / main / java / de / juplo / kafka / outbox / delivery / Watermarks.java

diff --git a/delivery/src/main/java/de/juplo/kafka/outbox/delivery/Watermarks.java b/delivery/src/main/java/de/juplo/kafka/outbox/delivery/Watermarks.java
index 8f071f9..4bd0c9e 100644 (file)
--- a/delivery/src/main/java/de/juplo/kafka/outbox/delivery/Watermarks.java
+++ b/delivery/src/main/java/de/juplo/kafka/outbox/delivery/Watermarks.java
@@ -3,19 +3,17 @@ package de.juplo.kafka.outbox.delivery;
 
 public class Watermarks
 {
-  private long[] watermarks = new long[0];
+  private final long[] watermarks;
 
 
-  public synchronized void set(int partition, long watermark)
+  public Watermarks(int partitions)
   {
-    if (partition >= watermarks.length)
-    {
-      long[] resized = new long[partition + 1];
-      for (int i = 0; i < watermarks.length; i++)
-        resized[i] = watermarks[i];
-      watermarks = resized;
-    }
+    watermarks = new long[partitions];
+  }
 
+
+  public synchronized void set(int partition, long watermark)
+  {
     watermarks[partition] = watermark;
   }
 
@@ -29,4 +27,20 @@ public class Watermarks
 
     return lowest;
   }
+
+  @Override
+  public String toString()
+  {
+    StringBuilder builder = new StringBuilder();
+    for (int i = 0; i < watermarks.length; i++)
+    {
+      builder.append("partition[");
+      builder.append(i);
+      builder.append("]=");
+      builder.append(watermarks[i]);
+      if (i != watermarks.length - 1)
+        builder.append(", ");
+    }
+    return builder.toString();
+  }
 }