diff --git a/src/main/java/io/numaproj/numaflow/shared/NackOptions.java b/src/main/java/io/numaproj/numaflow/shared/NackOptions.java index 12fb35ec..2cfcbca4 100644 --- a/src/main/java/io/numaproj/numaflow/shared/NackOptions.java +++ b/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 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(); } } diff --git a/src/main/proto/common/nack_options.proto b/src/main/proto/common/nack_options.proto index 28fd72d6..c86f109f 100644 --- a/src/main/proto/common/nack_options.proto +++ b/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 nack_map = 4; } diff --git a/src/test/java/io/numaproj/numaflow/shared/NackOptionsTest.java b/src/test/java/io/numaproj/numaflow/shared/NackOptionsTest.java index 0af26ab3..0351ac80 100644 --- a/src/test/java/io/numaproj/numaflow/shared/NackOptionsTest.java +++ b/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,7 +13,9 @@ public class NackOptionsTest { @Test public void toProto_allFields() { - NackOptions n = NackOptions.newBuilder().delay(500L).maxDeliveries(3).reason("retry").build(); + HashMap 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()); @@ -19,6 +23,8 @@ public void toProto_allFields() { 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 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