You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2021/06/23 13:07:52 UTC
[GitHub] [flink-statefun] igalshilman opened a new pull request #241: [FLINK-21308] Support delayed message cancellation
igalshilman opened a new pull request #241:
URL: https://github.com/apache/flink-statefun/pull/241
### This PR adds the ability to cancel delayed messages.
- see [FLINK-21308](https://issues.apache.org/jira/browse/FLINK-21308) for the detailed use case that drives this PR.
## High level changes
This PR introduces the following methods in the embedded SDK, and the corresponding remote SDKs (language specific flavors)
```java
/**
* Invokes another function with an input (associated with a {@code cancellationToken}),
* identified by the target function's {@link Address}, after a given delay.
*
* <p>Providing an id to a message, allows "unsending" this message later. ({@link
* #cancelDelayedMessage(String)}).
*
* @param delay the amount of delay before invoking the target function. Value needs to be >=
* 0.
* @param to the target function's address.
* @param message the input to provide for the delayed invocation.
* @param cancellationToken the non-empty, non-null, unique token to attach to this message, to be
* used for message cancellation. (see {@link #cancelDelayedMessage(String)}.)
*/
void sendAfter(Duration delay, Address to, Object message, String cancellationToken);
/**
* Cancel a delayed message (a message that was send via {@link #sendAfter(Duration, Address,
* Object, String)}).
*
* <p>NOTE: this is a best-effort operation, since the message might have been already delivered.
* If the message was delivered, this is a no-op operation.
*
* @param cancellationToken the id of the message to un-send.
*/
void cancelDelayedMessage(String cancellationToken);
```
* The semantic of the cancelation token is opaque to StateFun, it it's content and uniqueness is completely user defined.
* violating uniqueness constraint for two different methods will result in a runtime error.
* Once the delayed message has been dispatched, the cancellation token is forgotten.
### Internal Changes
## request reply protocol:
We attach the (optional) `cancelltion_token` to the delayed invocation in the request-reply protocol.
```protobuf
// DelayedInvocation represents a delayed remote function call with a target address, an argument
// and a delay in milliseconds, after which this message to be sent.
message DelayedInvocation {
// an optional cancellation token that can be used to request the "unsending" of a delayed message.
string cancellation_token = 10;
// the amount of milliseconds to wait before sending this message
int64 delay_in_ms = 1;
// the target address to send this message to
Address target = 2;
// the invocation argument
TypedValue argument = 3;
}
```
In addition, a new response message:
```protobuf
// DelayCancellation represents a single delayed-message cancellation request.
message DelayCancellation {
string cancellation_token = 1;
}
```
## State
We add an additional state handle (`delayed-message-index`) to keep track of the mapping between a `canceltion_token` and the absolute timestamp that this message needs to be dispatched at.
These changes are then wired throughout the SDK and the runtime to make the magic happen.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] tzulitai commented on a change in pull request #241: [FLINK-21308] Support delayed message cancellation
Posted by GitBox <gi...@apache.org>.
tzulitai commented on a change in pull request #241:
URL: https://github.com/apache/flink-statefun/pull/241#discussion_r662804963
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/functions/DelaySink.java
##########
@@ -64,35 +56,25 @@ void accept(Message message, long delayMillis) {
}
@Override
- public void onProcessingTime(InternalTimer<String, VoidNamespace> timer) throws Exception {
- final long triggerTimestamp = timer.getTimestamp();
- final Reductions reductions = reductionsSupplier.get();
-
- Iterable<Message> delayedMessages = delayedMessagesBuffer.getForTimestamp(triggerTimestamp);
- if (delayedMessages == null) {
- throw new IllegalStateException(
- "A delayed message timer was triggered with timestamp "
- + triggerTimestamp
- + ", but no messages were buffered for it.");
- }
- for (Message delayedMessage : delayedMessages) {
- if (thisPartition.contains(delayedMessage.target())) {
- reductions.enqueue(delayedMessage);
- } else {
- remoteSink.accept(delayedMessage);
- }
- }
- // we clear the delayedMessageBuffer *before* we process the enqueued local reductions, because
- // processing the envelops might actually trigger a delayed message to be sent with the same
- // @triggerTimestamp
- // so it would be re-enqueued into the delayedMessageBuffer.
- delayedMessagesBuffer.clearForTimestamp(triggerTimestamp);
- reductions.processEnvelopes();
+ public void onProcessingTime(InternalTimer<String, VoidNamespace> timer) {
Review comment:
👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] tzulitai commented on pull request #241: [FLINK-21308] Support delayed message cancellation
Posted by GitBox <gi...@apache.org>.
tzulitai commented on pull request #241:
URL: https://github.com/apache/flink-statefun/pull/241#issuecomment-876126960
Thanks @igalshilman! Merging this!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] tzulitai commented on a change in pull request #241: [FLINK-21308] Support delayed message cancellation
Posted by GitBox <gi...@apache.org>.
tzulitai commented on a change in pull request #241:
URL: https://github.com/apache/flink-statefun/pull/241#discussion_r662806624
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/functions/FlinkStateDelayedMessagesBuffer.java
##########
@@ -26,41 +33,121 @@
final class FlinkStateDelayedMessagesBuffer implements DelayedMessagesBuffer {
static final String BUFFER_STATE_NAME = "delayed-messages-buffer";
+ static final String INDEX_STATE_NAME = "delayed-message-index";
private final InternalListState<String, Long, Message> bufferState;
+ private final MapState<String, Long> messageIdToTimestamp;
@Inject
FlinkStateDelayedMessagesBuffer(
- @Label("delayed-messages-buffer-state")
- InternalListState<String, Long, Message> bufferState) {
+ @Label("delayed-messages-buffer-state") InternalListState<String, Long, Message> bufferState,
+ @Label("delayed-message-index") MapState<String, Long> messageIdToTimestamp) {
this.bufferState = Objects.requireNonNull(bufferState);
+ this.messageIdToTimestamp = Objects.requireNonNull(messageIdToTimestamp);
}
@Override
- public void add(Message message, long untilTimestamp) {
- bufferState.setCurrentNamespace(untilTimestamp);
+ public void forEachMessageAt(long timestamp, Consumer<Message> fn) {
try {
- bufferState.add(message);
+ forEachMessageThrows(timestamp, fn);
} catch (Exception e) {
- throw new RuntimeException("Error adding delayed message to state buffer: " + message, e);
+ throw new IllegalStateException(e);
}
}
@Override
- public Iterable<Message> getForTimestamp(long timestamp) {
- bufferState.setCurrentNamespace(timestamp);
-
+ public OptionalLong removeMessageByCancellationToken(String token) {
try {
- return bufferState.get();
+ return remove(token);
} catch (Exception e) {
- throw new RuntimeException(
- "Error accessing delayed message in state buffer for timestamp: " + timestamp, e);
+ throw new IllegalStateException("Failed clearing a message with id " + token, e);
}
}
@Override
- public void clearForTimestamp(long timestamp) {
+ public void add(Message message, long untilTimestamp) {
+ try {
+ addThrows(message, untilTimestamp);
+ } catch (Exception e) {
+ throw new RuntimeException("Error adding delayed message to state buffer: " + message, e);
+ }
+ }
+
+ // -----------------------------------------------------------------------------------------------------
+ // Internal
+ // -----------------------------------------------------------------------------------------------------
+
+ private void forEachMessageThrows(long timestamp, Consumer<Message> fn) throws Exception {
bufferState.setCurrentNamespace(timestamp);
+ for (Message message : bufferState.get()) {
+ removeMessageIdMapping(message);
+ fn.accept(message);
+ }
+ bufferState.clear();
+ }
+
+ private void addThrows(Message message, long untilTimestamp) throws Exception {
+ bufferState.setCurrentNamespace(untilTimestamp);
+ bufferState.add(message);
+ Optional<String> maybeId = message.cancellationToken();
+ if (!maybeId.isPresent()) {
+ return;
+ }
+ String messageId = maybeId.get();
+ @Nullable Long previousTimestamp = messageIdToTimestamp.get(messageId);
+ if (previousTimestamp != null) {
+ throw new IllegalStateException(
+ "Trying to associate a message with id "
+ + messageId
+ + " and timestamp "
+ + untilTimestamp
+ + ", but a message with the same id exists and with a timestamp "
Review comment:
Same here: `id` --> `cancellation token` in the error messages.
We seem to refer to these as cancellation tokens on the API surface.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] tzulitai closed pull request #241: [FLINK-21308] Support delayed message cancellation
Posted by GitBox <gi...@apache.org>.
tzulitai closed pull request #241:
URL: https://github.com/apache/flink-statefun/pull/241
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] igalshilman commented on pull request #241: [FLINK-21308] Support delayed message cancellation
Posted by GitBox <gi...@apache.org>.
igalshilman commented on pull request #241:
URL: https://github.com/apache/flink-statefun/pull/241#issuecomment-875580535
@tzulitai Thanks for the review! I've squashed the previous history into a single commit, as the history wasn't that important. and the last 4 commits are the result of our conversation earlier today.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] tzulitai commented on a change in pull request #241: [FLINK-21308] Support delayed message cancellation
Posted by GitBox <gi...@apache.org>.
tzulitai commented on a change in pull request #241:
URL: https://github.com/apache/flink-statefun/pull/241#discussion_r662805939
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/functions/FlinkStateDelayedMessagesBuffer.java
##########
@@ -26,41 +33,121 @@
final class FlinkStateDelayedMessagesBuffer implements DelayedMessagesBuffer {
static final String BUFFER_STATE_NAME = "delayed-messages-buffer";
+ static final String INDEX_STATE_NAME = "delayed-message-index";
private final InternalListState<String, Long, Message> bufferState;
+ private final MapState<String, Long> messageIdToTimestamp;
@Inject
FlinkStateDelayedMessagesBuffer(
- @Label("delayed-messages-buffer-state")
- InternalListState<String, Long, Message> bufferState) {
+ @Label("delayed-messages-buffer-state") InternalListState<String, Long, Message> bufferState,
+ @Label("delayed-message-index") MapState<String, Long> messageIdToTimestamp) {
this.bufferState = Objects.requireNonNull(bufferState);
+ this.messageIdToTimestamp = Objects.requireNonNull(messageIdToTimestamp);
}
@Override
- public void add(Message message, long untilTimestamp) {
- bufferState.setCurrentNamespace(untilTimestamp);
+ public void forEachMessageAt(long timestamp, Consumer<Message> fn) {
try {
- bufferState.add(message);
+ forEachMessageThrows(timestamp, fn);
} catch (Exception e) {
- throw new RuntimeException("Error adding delayed message to state buffer: " + message, e);
+ throw new IllegalStateException(e);
}
}
@Override
- public Iterable<Message> getForTimestamp(long timestamp) {
- bufferState.setCurrentNamespace(timestamp);
-
+ public OptionalLong removeMessageByCancellationToken(String token) {
try {
- return bufferState.get();
+ return remove(token);
} catch (Exception e) {
- throw new RuntimeException(
- "Error accessing delayed message in state buffer for timestamp: " + timestamp, e);
+ throw new IllegalStateException("Failed clearing a message with id " + token, e);
Review comment:
Maybe change `id` in the error message to `cancellation token`, to be consistent and not confuse users.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-statefun] tzulitai commented on a change in pull request #241: [FLINK-21308] Support delayed message cancellation
Posted by GitBox <gi...@apache.org>.
tzulitai commented on a change in pull request #241:
URL: https://github.com/apache/flink-statefun/pull/241#discussion_r662801961
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/reqreply/RequestReplyFunction.java
##########
@@ -241,6 +242,16 @@ private void handleOutgoingDelayedMessages(Context context, InvocationResponse i
}
}
+ private void handleDelayedCancellations(InternalContext context, InvocationResponse result) {
+ for (FromFunction.DelayCancellation delayCancellation :
+ result.getOutgoingDelayCancellationsList()) {
+ String token = delayCancellation.getCancellationToken();
+ if (!token.isEmpty()) {
+ context.cancelDelayedMessage(delayCancellation.getCancellationToken());
Review comment:
I think this might even be a common case.
Say, for the same invocation batch with a given key, an earlier message triggers a delayed message that is cancellable. Maybe as a timer to purge state etc. in the absence of a follow up event.
However, within that same batch, the follow up event is processed and users immediately wants to cancel the delayed message. We can maybe optimize this scenario?
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/reqreply/RequestReplyFunction.java
##########
@@ -241,6 +242,16 @@ private void handleOutgoingDelayedMessages(Context context, InvocationResponse i
}
}
+ private void handleDelayedCancellations(InternalContext context, InvocationResponse result) {
+ for (FromFunction.DelayCancellation delayCancellation :
+ result.getOutgoingDelayCancellationsList()) {
+ String token = delayCancellation.getCancellationToken();
+ if (!token.isEmpty()) {
+ context.cancelDelayedMessage(delayCancellation.getCancellationToken());
Review comment:
Assume the following scenario:
Within the same `FromFunction` response from a remote function, there is a delayed message with cancellation token `foobar`. At the same time, there is also a cancellation token for `foobar`.
I'm wondering if we want to be diligent in this scenario and eagerly "cancel" out the delayed message, without going through a roundtrip of adding it to the buffer / index state, just to delete it almost imediately.
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/functions/FlinkStateDelayedMessagesBuffer.java
##########
@@ -26,41 +33,121 @@
final class FlinkStateDelayedMessagesBuffer implements DelayedMessagesBuffer {
static final String BUFFER_STATE_NAME = "delayed-messages-buffer";
+ static final String INDEX_STATE_NAME = "delayed-message-index";
private final InternalListState<String, Long, Message> bufferState;
+ private final MapState<String, Long> messageIdToTimestamp;
@Inject
FlinkStateDelayedMessagesBuffer(
- @Label("delayed-messages-buffer-state")
- InternalListState<String, Long, Message> bufferState) {
+ @Label("delayed-messages-buffer-state") InternalListState<String, Long, Message> bufferState,
+ @Label("delayed-message-index") MapState<String, Long> messageIdToTimestamp) {
this.bufferState = Objects.requireNonNull(bufferState);
+ this.messageIdToTimestamp = Objects.requireNonNull(messageIdToTimestamp);
}
@Override
- public void add(Message message, long untilTimestamp) {
- bufferState.setCurrentNamespace(untilTimestamp);
+ public void forEachMessageAt(long timestamp, Consumer<Message> fn) {
try {
- bufferState.add(message);
+ forEachMessageThrows(timestamp, fn);
} catch (Exception e) {
- throw new RuntimeException("Error adding delayed message to state buffer: " + message, e);
+ throw new IllegalStateException(e);
}
}
@Override
- public Iterable<Message> getForTimestamp(long timestamp) {
- bufferState.setCurrentNamespace(timestamp);
-
+ public OptionalLong removeMessageByCancellationToken(String token) {
try {
- return bufferState.get();
+ return remove(token);
} catch (Exception e) {
- throw new RuntimeException(
- "Error accessing delayed message in state buffer for timestamp: " + timestamp, e);
+ throw new IllegalStateException("Failed clearing a message with id " + token, e);
}
}
@Override
- public void clearForTimestamp(long timestamp) {
+ public void add(Message message, long untilTimestamp) {
+ try {
+ addThrows(message, untilTimestamp);
+ } catch (Exception e) {
+ throw new RuntimeException("Error adding delayed message to state buffer: " + message, e);
+ }
+ }
+
+ // -----------------------------------------------------------------------------------------------------
+ // Internal
+ // -----------------------------------------------------------------------------------------------------
+
+ private void forEachMessageThrows(long timestamp, Consumer<Message> fn) throws Exception {
bufferState.setCurrentNamespace(timestamp);
+ for (Message message : bufferState.get()) {
+ removeMessageIdMapping(message);
+ fn.accept(message);
+ }
+ bufferState.clear();
+ }
+
+ private void addThrows(Message message, long untilTimestamp) throws Exception {
+ bufferState.setCurrentNamespace(untilTimestamp);
+ bufferState.add(message);
+ Optional<String> maybeId = message.cancellationToken();
+ if (!maybeId.isPresent()) {
+ return;
+ }
+ String messageId = maybeId.get();
+ @Nullable Long previousTimestamp = messageIdToTimestamp.get(messageId);
+ if (previousTimestamp != null) {
+ throw new IllegalStateException(
+ "Trying to associate a message with id "
+ + messageId
+ + " and timestamp "
+ + untilTimestamp
+ + ", but a message with the same id exists and with a timestamp "
Review comment:
nit: excessive "and"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org