feat: introduce a generic map for per message nack options

src/main/java/io/numaproj/numaflow/shared/NackOptions.java

@@ -3,6 +3,8 @@
 import lombok.Builder;
 import lombok.Getter;
 
+import java.util.Map;
+
 /**
  * NackOptions carries per-message redelivery options for a negative acknowledgement (nack).
  * All fields are optional; a null value means unset.
@@ -16,6 +18,8 @@ public class NackOptions {
     private final Integer maxDeliveries;
     /** human-readable reason for the nack. */
     private final String reason;
+    /** generic values passed as nack options */
+    private final Map<String, String> nackMap;
 
     /** Converts to the outgoing proto type, setting only the fields that are present. */
     public common.NackOptionsOuterClass.NackOptions toProto() {
@@ -30,6 +34,9 @@ public common.NackOptionsOuterClass.NackOptions toProto() {
         if (reason != null) {
             b.setReason(reason);
         }
+        if (nackMap != null) {
+            b.putAllNackMap(nackMap);
+        }
         return b.build();
     }
 
@@ -48,6 +55,9 @@ public static NackOptions fromProto(common.NackOptionsOuterClass.NackOptions p)
         if (p.hasReason()) {
             b.reason(p.getReason());
         }
+        if (!p.getNackMapMap().isEmpty()) {
+            b.nackMap(p.getNackMapMap());
+        }
         return b.build();
     }
 }

src/main/proto/common/nack_options.proto

@@ -6,4 +6,5 @@ message NackOptions {
   optional string reason = 1;
   optional uint32 max_deliveries = 2;
   optional uint64 delay = 3;
+  map<string, string> nack_map = 4;
 }

src/test/java/io/numaproj/numaflow/shared/NackOptionsTest.java

@@ -2,6 +2,8 @@
 
 import org.junit.Test;
 
+import java.util.HashMap;
+
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNull;
@@ -11,14 +13,18 @@ public class NackOptionsTest {
 
     @Test
     public void toProto_allFields() {
-        NackOptions n = NackOptions.newBuilder().delay(500L).maxDeliveries(3).reason("retry").build();
+        HashMap<String, String> nackMap = new HashMap<>();
+        nackMap.put("key", "value");
+        NackOptions n = NackOptions.newBuilder().delay(500L).maxDeliveries(3).reason("retry").nackMap(nackMap).build();
         common.NackOptionsOuterClass.NackOptions p = n.toProto();
         assertTrue(p.hasDelay());
         assertEquals(500L, p.getDelay());
         assertTrue(p.hasMaxDeliveries());
         assertEquals(3, p.getMaxDeliveries());
         assertTrue(p.hasReason());
         assertEquals("retry", p.getReason());
+        assertFalse(p.getNackMapMap().isEmpty());
+        assertEquals("value", p.getNackMapMap().get("key"));
     }
 
     @Test
@@ -32,12 +38,15 @@ public void toProto_partialFields() {
 
     @Test
     public void fromProto_roundTrip() {
+        HashMap<String, String> nackMap = new HashMap<>();
+        nackMap.put("key", "value");
         common.NackOptionsOuterClass.NackOptions p = common.NackOptionsOuterClass.NackOptions.newBuilder()
-                .setDelay(500L).setMaxDeliveries(3).setReason("retry").build();
+                .setDelay(500L).setMaxDeliveries(3).setReason("retry").putAllNackMap(nackMap).build();
         NackOptions n = NackOptions.fromProto(p);
         assertEquals(Long.valueOf(500L), n.getDelay());
         assertEquals(Integer.valueOf(3), n.getMaxDeliveries());
         assertEquals("retry", n.getReason());
+        assertEquals(nackMap, n.getNackMap());
     }
 
     @Test