Skip to content

Commit 7f129ed

Browse files
vaibhavtiwari33vaibhavtiwari33
authored
feat: end to end wiring for per message nack (#220)
Signed-off-by: Vaibhav Tiwari <vaibhav.tiwari33@gmail.com>
1 parent 0c3b01c commit 7f129ed

30 files changed

Lines changed: 850 additions & 59 deletions

File tree

‎examples/src/main/java/io/numaproj/numaflow/examples/source/simple/SimpleSource.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22

33
import io.numaproj.numaflow.sourcer.AckRequest;
44
import io.numaproj.numaflow.sourcer.Message;
5+
import io.numaproj.numaflow.sourcer.NackOffset;
56
import io.numaproj.numaflow.sourcer.NackRequest;
67
import io.numaproj.numaflow.sourcer.Offset;
78
import io.numaproj.numaflow.sourcer.OutputObserver;
@@ -86,8 +87,8 @@ public void ack(AckRequest request) {
8687
@Override
8788
public void nack(NackRequest request) {
8889
// put them to nacked offsets so that they will be retried immediately.
89-
for (Offset offset : request.getOffsets()) {
90-
Integer decoded_offset = ByteBuffer.wrap(offset.getValue()).getInt();
90+
for (NackOffset offset : request.getOffsets()) {
91+
Integer decoded_offset = ByteBuffer.wrap(offset.getOffset().getValue()).getInt();
9192
yetToBeAcked.remove(decoded_offset);
9293
nacked.put(decoded_offset, true);
9394
readIndex.decrementAndGet();

‎src/main/java/io/numaproj/numaflow/batchmapper/Message.java‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,17 @@
11
package io.numaproj.numaflow.batchmapper;
22

3+
import io.numaproj.numaflow.shared.NackOptions;
34
import lombok.Getter;
45

56
/** Message is used to wrap the data returned by Mapper. */
67
@Getter
78
public class Message {
89
private static final String[] DROP_TAGS = {"U+005C__DROP__"};
10+
private static final String[] NACK_TAGS = {"U+005C__NACK__"};
911
private final String[] keys;
1012
private final byte[] value;
1113
private final String[] tags;
14+
private final NackOptions nackOptions;
1215

1316
/**
1417
* used to create Message with value, keys and tags(used for conditional forwarding)
@@ -18,10 +21,15 @@ public class Message {
1821
* @param tags message tags which will be used for conditional forwarding
1922
*/
2023
public Message(byte[] value, String[] keys, String[] tags) {
24+
this(value, keys, tags, (NackOptions) null);
25+
}
26+
27+
private Message(byte[] value, String[] keys, String[] tags, NackOptions nackOptions) {
2128
// defensive copy - once the Message is created, the caller should not be able to modify it.
2229
this.keys = keys == null ? null : keys.clone();
2330
this.value = value == null ? null : value.clone();
2431
this.tags = tags == null ? null : tags.clone();
32+
this.nackOptions = nackOptions;
2533
}
2634

2735
/**
@@ -51,4 +59,14 @@ public Message(byte[] value, String[] keys) {
5159
public static Message toDrop() {
5260
return new Message(new byte[0], null, DROP_TAGS);
5361
}
62+
63+
/**
64+
* creates a Message that negatively acknowledges the input message, requesting redelivery.
65+
*
66+
* @param nackOptions optional redelivery options (may be null)
67+
* @return the Message which will be nacked
68+
*/
69+
public static Message toNack(NackOptions nackOptions) {
70+
return new Message(new byte[0], null, NACK_TAGS, nackOptions);
71+
}
5472
}

‎src/main/java/io/numaproj/numaflow/batchmapper/Service.java‎

Lines changed: 9 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -137,18 +137,15 @@ private void buildAndStreamResponse(
137137
responses.getItems().forEach(message -> {
138138
List<MapOuterClass.MapResponse.Result> mapResponseResult = new ArrayList<>();
139139
message.getItems().forEach(res -> {
140-
mapResponseResult.add(
141-
MapOuterClass.MapResponse.Result
142-
.newBuilder()
143-
.setValue(res.getValue()
144-
== null ? ByteString.EMPTY : ByteString.copyFrom(
145-
res.getValue()))
146-
.addAllKeys(res.getKeys()
147-
== null ? new ArrayList<>() : Arrays.asList(res.getKeys()))
148-
.addAllTags(res.getTags()
149-
== null ? new ArrayList<>() : Arrays.asList(res.getTags()))
150-
.build()
151-
);
140+
MapOuterClass.MapResponse.Result.Builder resultBuilder = MapOuterClass.MapResponse.Result
141+
.newBuilder()
142+
.setValue(res.getValue() == null ? ByteString.EMPTY : ByteString.copyFrom(res.getValue()))
143+
.addAllKeys(res.getKeys() == null ? new ArrayList<>() : Arrays.asList(res.getKeys()))
144+
.addAllTags(res.getTags() == null ? new ArrayList<>() : Arrays.asList(res.getTags()));
145+
if (res.getNackOptions() != null) {
146+
resultBuilder.setNackOptions(res.getNackOptions().toProto());
147+
}
148+
mapResponseResult.add(resultBuilder.build());
152149
});
153150
MapOuterClass.MapResponse singleRequestResponse = MapOuterClass.MapResponse
154151
.newBuilder()

‎src/main/java/io/numaproj/numaflow/mapper/MapperActor.java‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -87,16 +87,19 @@ private MapOuterClass.MapResponse buildResponse(MessageList messageList, String
8787
.newBuilder();
8888

8989
messageList.getMessages().forEach(message -> {
90-
responseBuilder.addResults(MapOuterClass.MapResponse.Result.newBuilder()
90+
MapOuterClass.MapResponse.Result.Builder resultBuilder = MapOuterClass.MapResponse.Result.newBuilder()
9191
.setValue(message.getValue() == null ? ByteString.EMPTY : ByteString.copyFrom(
9292
message.getValue()))
9393
.addAllKeys(message.getKeys()
9494
== null ? new ArrayList<>() : Arrays.asList(message.getKeys()))
9595
.addAllTags(message.getTags()
9696
== null ? new ArrayList<>() : Arrays.asList(message.getTags()))
9797
.setMetadata(message.getUserMetadata()
98-
== null ? MetadataOuterClass.Metadata.getDefaultInstance() : message.getUserMetadata().toProto())
99-
.build());
98+
== null ? MetadataOuterClass.Metadata.getDefaultInstance() : message.getUserMetadata().toProto());
99+
if (message.getNackOptions() != null) {
100+
resultBuilder.setNackOptions(message.getNackOptions().toProto());
101+
}
102+
responseBuilder.addResults(resultBuilder.build());
100103
});
101104
return responseBuilder.setId(ID).build();
102105
}

‎src/main/java/io/numaproj/numaflow/mapper/Message.java‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package io.numaproj.numaflow.mapper;
22

3+
import io.numaproj.numaflow.shared.NackOptions;
34
import io.numaproj.numaflow.shared.UserMetadata;
45
import lombok.Getter;
56

@@ -11,10 +12,12 @@
1112
@Getter
1213
public class Message {
1314
private static final String[] DROP_TAGS = {"U+005C__DROP__"};
15+
private static final String[] NACK_TAGS = {"U+005C__NACK__"};
1416
private final String[] keys;
1517
private final byte[] value;
1618
private final String[] tags;
1719
private final UserMetadata userMetadata;
20+
private final NackOptions nackOptions;
1821

1922
/**
2023
* used to create Message with value, keys, tags(used for conditional forwarding) and userMetadata
@@ -25,12 +28,17 @@ public class Message {
2528
* @param userMetadata user metadata, this is used to pass user defined metadata to the next vertex
2629
*/
2730
public Message(byte[] value, String[] keys, String[] tags, UserMetadata userMetadata) {
31+
this(value, keys, tags, userMetadata, null);
32+
}
33+
34+
private Message(byte[] value, String[] keys, String[] tags, UserMetadata userMetadata, NackOptions nackOptions) {
2835
// defensive copy - once the Message is created, the caller should not be able to modify it.
2936
this.keys = keys == null ? null : keys.clone();
3037
this.value = value == null ? null : value.clone();
3138
this.tags = tags == null ? null : tags.clone();
3239
// Copy the data using copy constructor to prevent mutation
3340
this.userMetadata = userMetadata == null ? null : new UserMetadata(userMetadata);
41+
this.nackOptions = nackOptions;
3442
}
3543

3644
/**
@@ -71,4 +79,14 @@ public Message(byte[] value, String[] keys, String[] tags) {
7179
public static Message toDrop() {
7280
return new Message(new byte[0], null, DROP_TAGS, null);
7381
}
82+
83+
/**
84+
* creates a Message that negatively acknowledges the input message, requesting redelivery.
85+
*
86+
* @param nackOptions optional redelivery options (may be null)
87+
* @return the Message which will be nacked
88+
*/
89+
public static Message toNack(NackOptions nackOptions) {
90+
return new Message(new byte[0], null, NACK_TAGS, null, nackOptions);
91+
}
7492
}

‎src/main/java/io/numaproj/numaflow/mapstreamer/Message.java‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,17 @@
11
package io.numaproj.numaflow.mapstreamer;
22

3+
import io.numaproj.numaflow.shared.NackOptions;
34
import lombok.Getter;
45

56
/** Message is used to wrap the data returned by MapStreamer. */
67
@Getter
78
public class Message {
89
private static final String[] DROP_TAGS = {"U+005C__DROP__"};
10+
private static final String[] NACK_TAGS = {"U+005C__NACK__"};
911
private final String[] keys;
1012
private final byte[] value;
1113
private final String[] tags;
14+
private final NackOptions nackOptions;
1215

1316
/**
1417
* used to create Message with value, keys and tags(used for conditional forwarding)
@@ -18,10 +21,15 @@ public class Message {
1821
* @param tags message tags which will be used for conditional forwarding
1922
*/
2023
public Message(byte[] value, String[] keys, String[] tags) {
24+
this(value, keys, tags, (NackOptions) null);
25+
}
26+
27+
private Message(byte[] value, String[] keys, String[] tags, NackOptions nackOptions) {
2128
// defensive copy - once the Message is created, the caller should not be able to modify it.
2229
this.keys = keys == null ? null : keys.clone();
2330
this.value = value == null ? null : value.clone();
2431
this.tags = tags == null ? null : tags.clone();
32+
this.nackOptions = nackOptions;
2533
}
2634

2735
/**
@@ -51,4 +59,14 @@ public Message(byte[] value, String[] keys) {
5159
public static Message toDrop() {
5260
return new Message(new byte[0], null, DROP_TAGS);
5361
}
62+
63+
/**
64+
* creates a Message that negatively acknowledges the input message, requesting redelivery.
65+
*
66+
* @param nackOptions optional redelivery options (may be null)
67+
* @return the Message which will be nacked
68+
*/
69+
public static Message toNack(NackOptions nackOptions) {
70+
return new Message(new byte[0], null, NACK_TAGS, nackOptions);
71+
}
5472
}

‎src/main/java/io/numaproj/numaflow/mapstreamer/OutputObserverImpl.java‎

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -29,17 +29,17 @@ public void send(Message message) {
2929
if (message == null) {
3030
return;
3131
}
32+
MapOuterClass.MapResponse.Result.Builder resultBuilder = MapOuterClass.MapResponse.Result.newBuilder()
33+
.setValue(message.getValue() == null ? ByteString.EMPTY : ByteString.copyFrom(message.getValue()))
34+
.addAllKeys(message.getKeys() == null ? new ArrayList<>() : Arrays.asList(message.getKeys()))
35+
.addAllTags(message.getTags() == null ? new ArrayList<>() : Arrays.asList(message.getTags()));
36+
if (message.getNackOptions() != null) {
37+
resultBuilder.setNackOptions(message.getNackOptions().toProto());
38+
}
3239
MapOuterClass.MapResponse response = MapOuterClass.MapResponse.newBuilder()
3340
.setId(requestID)
34-
.addResults(MapOuterClass.MapResponse.Result.newBuilder()
35-
.setValue(
36-
message.getValue() == null ? ByteString.EMPTY : ByteString.copyFrom(
37-
message.getValue()))
38-
.addAllKeys(message.getKeys()
39-
== null ? new ArrayList<>() : Arrays.asList(message.getKeys()))
40-
.addAllTags(message.getTags()
41-
== null ? new ArrayList<>() : Arrays.asList(message.getTags()))
42-
.build()).build();
41+
.addResults(resultBuilder.build())
42+
.build();
4343
supervisorActor.tell(response, ActorRef.noSender());
4444
}
4545

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

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
package io.numaproj.numaflow.shared;
2+
3+
import lombok.Builder;
4+
import lombok.Getter;
5+
6+
/**
7+
* NackOptions carries per-message redelivery options for a negative acknowledgement (nack).
8+
* All fields are optional; a null value means unset.
9+
*/
10+
@Getter
11+
@Builder(builderMethodName = "newBuilder")
12+
public class NackOptions {
13+
/** redelivery delay in milliseconds. */
14+
private final Long delay;
15+
/** maximum number of redelivery attempts. */
16+
private final Integer maxDeliveries;
17+
/** human-readable reason for the nack. */
18+
private final String reason;
19+
20+
/** Converts to the outgoing proto type, setting only the fields that are present. */
21+
public common.NackOptionsOuterClass.NackOptions toProto() {
22+
common.NackOptionsOuterClass.NackOptions.Builder b =
23+
common.NackOptionsOuterClass.NackOptions.newBuilder();
24+
if (delay != null) {
25+
b.setDelay(delay);
26+
}
27+
if (maxDeliveries != null) {
28+
b.setMaxDeliveries(maxDeliveries);
29+
}
30+
if (reason != null) {
31+
b.setReason(reason);
32+
}
33+
return b.build();
34+
}
35+
36+
/** Converts from the incoming proto type. Returns null for null input. */
37+
public static NackOptions fromProto(common.NackOptionsOuterClass.NackOptions p) {
38+
if (p == null) {
39+
return null;
40+
}
41+
NackOptionsBuilder b = NackOptions.newBuilder();
42+
if (p.hasDelay()) {
43+
b.delay(p.getDelay());
44+
}
45+
if (p.hasMaxDeliveries()) {
46+
b.maxDeliveries(p.getMaxDeliveries());
47+
}
48+
if (p.hasReason()) {
49+
b.reason(p.getReason());
50+
}
51+
return b.build();
52+
}
53+
}

‎src/main/java/io/numaproj/numaflow/sinker/Response.java‎

Lines changed: 20 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package io.numaproj.numaflow.sinker;
22

3+
import io.numaproj.numaflow.shared.NackOptions;
34
import lombok.AccessLevel;
45
import lombok.AllArgsConstructor;
56
import lombok.Getter;
@@ -20,6 +21,8 @@ public class Response {
2021
private final byte[] serveResponse;
2122
private final Boolean onSuccess;
2223
private final Message onSuccessMessage;
24+
private final Boolean nack;
25+
private final NackOptions nackOptions;
2326

2427
/**
2528
* Static method to create response for successful message processing.
@@ -28,7 +31,7 @@ public class Response {
2831
* @return Response object with success status
2932
*/
3033
public static Response responseOK(String id) {
31-
return new Response(id, true, null, false, false, null, false, null);
34+
return new Response(id, true, null, false, false, null, false, null, false, null);
3235
}
3336

3437
/**
@@ -39,7 +42,7 @@ public static Response responseOK(String id) {
3942
* @return Response object with failure status and error message
4043
*/
4144
public static Response responseFailure(String id, String errMsg) {
42-
return new Response(id, false, errMsg, false, false, null, false, null);
45+
return new Response(id, false, errMsg, false, false, null, false, null, false, null);
4346
}
4447

4548
/**
@@ -50,7 +53,7 @@ public static Response responseFailure(String id, String errMsg) {
5053
* @return Response object with fallback status
5154
*/
5255
public static Response responseFallback(String id) {
53-
return new Response(id, false, null, true, false, null, false, null);
56+
return new Response(id, false, null, true, false, null, false, null, false, null);
5457
}
5558

5659
/**
@@ -63,7 +66,7 @@ public static Response responseFallback(String id) {
6366
* @return Response object with serve status and serve response
6467
*/
6568
public static Response responseServe(String id, byte[] serveResponse) {
66-
return new Response(id, false, null, false, true, serveResponse, false, null);
69+
return new Response(id, false, null, false, true, serveResponse, false, null, false, null);
6770
}
6871

6972
/**
@@ -76,6 +79,18 @@ public static Response responseServe(String id, byte[] serveResponse) {
7679
* @return Response object with onSuccess status and onSuccess message
7780
*/
7881
public static Response responseOnSuccess(String id, Message onSuccessMessage) {
79-
return new Response(id, false, null, false, false, null, true, onSuccessMessage);
82+
return new Response(id, false, null, false, false, null, true, onSuccessMessage, false, null);
83+
}
84+
85+
/**
86+
* Static method to create a nack response, indicating the message should be negatively
87+
* acknowledged and redelivered. nackOptions may be null.
88+
*
89+
* @param id id of the message
90+
* @param nackOptions optional redelivery options
91+
* @return Response object with nack status
92+
*/
93+
public static Response responseNack(String id, NackOptions nackOptions) {
94+
return new Response(id, false, null, false, false, null, false, null, true, nackOptions);
8095
}
8196
}

‎src/main/java/io/numaproj/numaflow/sinker/Service.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -161,6 +161,14 @@ private SinkOuterClass.SinkResponse.Result buildResult(Response response) {
161161
.setStatus(SinkOuterClass.Status.ON_SUCCESS)
162162
.setOnSuccessMsg(Message.toProto(response.getOnSuccessMessage()))
163163
.build();
164+
} else if (response.getNack() != null && response.getNack()) {
165+
SinkOuterClass.SinkResponse.Result.Builder b = SinkOuterClass.SinkResponse.Result.newBuilder()
166+
.setId(response.getId() == null ? "" : response.getId())
167+
.setStatus(SinkOuterClass.Status.NACK);
168+
if (response.getNackOptions() != null) {
169+
b.setNackOptions(response.getNackOptions().toProto());
170+
}
171+
return b.build();
164172
} else {
165173
// FIXME: Return error when error message is not set?
166174
return SinkOuterClass.SinkResponse.Result.newBuilder()

0 commit comments

Comments
 (0)