diff --git a/statefun-e2e-tests/statefun-smoke-e2e-common/src/main/java/org/apache/flink/statefun/e2e/smoke/SmokeRunnerParameters.java b/statefun-e2e-tests/statefun-smoke-e2e-common/src/main/java/org/apache/flink/statefun/e2e/smoke/SmokeRunnerParameters.java
index 2f63a5d..ef33fd0 100644
--- a/statefun-e2e-tests/statefun-smoke-e2e-common/src/main/java/org/apache/flink/statefun/e2e/smoke/SmokeRunnerParameters.java
+++ b/statefun-e2e-tests/statefun-smoke-e2e-common/src/main/java/org/apache/flink/statefun/e2e/smoke/SmokeRunnerParameters.java
@@ -35,6 +35,7 @@
   private double stateModificationsPr = 0.4;
   private double sendPr = 0.9;
   private double sendAfterPr = 0.1;
+  private double sendAfterWithCancellationPr = 0.1;
   private double asyncSendPr = 0.1;
   private double noopPr = 0.2;
   private double sendEgressPr = 0.03;
@@ -42,6 +43,8 @@
   private String verificationServerHost = "localhost";
   private int verificationServerPort = 5050;
   private boolean isAsyncOpSupported = false;
+  private boolean isDelayCancellationOpSupported = false;
+
   private long randomGeneratorSeed = System.nanoTime();
 
   /** Creates an instance of ModuleParameters from a key-value map. */
@@ -108,6 +111,14 @@
     return sendAfterPr;
   }
 
+  public boolean isDelayCancellationOpSupported() {
+    return isDelayCancellationOpSupported;
+  }
+
+  public void setDelayCancellationOpSupported(boolean delayCancellationOpSupported) {
+    isDelayCancellationOpSupported = delayCancellationOpSupported;
+  }
+
   public void setSendAfterPr(double sendAfterPr) {
     this.sendAfterPr = sendAfterPr;
   }
@@ -160,6 +171,14 @@
     this.verificationServerPort = verificationServerPort;
   }
 
+  public double getSendAfterWithCancellationPr() {
+    return sendAfterWithCancellationPr;
+  }
+
+  public void setSendAfterWithCancellationPr(double sendAfterWithCancellationPr) {
+    this.sendAfterWithCancellationPr = sendAfterWithCancellationPr;
+  }
+
   public boolean isAsyncOpSupported() {
     return isAsyncOpSupported;
   }
diff --git a/statefun-e2e-tests/statefun-smoke-e2e-common/src/main/protobuf/commands.proto b/statefun-e2e-tests/statefun-smoke-e2e-common/src/main/protobuf/commands.proto
index e3912d1..8bcd4cc 100644
--- a/statefun-e2e-tests/statefun-smoke-e2e-common/src/main/protobuf/commands.proto
+++ b/statefun-e2e-tests/statefun-smoke-e2e-common/src/main/protobuf/commands.proto
@@ -42,6 +42,11 @@
   message SendAfter {
     int32 target = 1;
     Commands commands = 2;
+    string cancellation_token = 3;
+  }
+  message CancelSendAfter {
+     int32 target = 1;
+     string cancellation_token = 2;
   }
   message SendEgress {
   }
@@ -60,6 +65,7 @@
     SendEgress send_egress = 4;
     AsyncOperation async_operation = 5;
     Verify verify = 6;
+    CancelSendAfter cancel_send_after = 7;
   }
 }
 
diff --git a/statefun-e2e-tests/statefun-smoke-e2e-driver/src/main/java/org/apache/flink/statefun/e2e/smoke/driver/CommandGenerator.java b/statefun-e2e-tests/statefun-smoke-e2e-driver/src/main/java/org/apache/flink/statefun/e2e/smoke/driver/CommandGenerator.java
index fa399ee..607ac61 100644
--- a/statefun-e2e-tests/statefun-smoke-e2e-driver/src/main/java/org/apache/flink/statefun/e2e/smoke/driver/CommandGenerator.java
+++ b/statefun-e2e-tests/statefun-smoke-e2e-driver/src/main/java/org/apache/flink/statefun/e2e/smoke/driver/CommandGenerator.java
@@ -23,6 +23,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
+import java.util.UUID;
 import java.util.function.Supplier;
 import org.apache.commons.math3.distribution.EnumeratedDistribution;
 import org.apache.commons.math3.random.RandomGenerator;
@@ -85,9 +86,13 @@
                 create(new SendAfterGen(), parameters.getSendAfterPr()),
                 create(new Noop(), parameters.getNoopPr()),
                 create(new SendEgress(), parameters.getSendEgressPr())));
+
     if (parameters.isAsyncOpSupported()) {
       list.add(create(new SendAsyncOp(), parameters.getAsyncSendPr()));
     }
+    if (parameters.isDelayCancellationOpSupported()) {
+      list.add(create(new SendAfterCancellationGen(), parameters.getSendAfterWithCancellationPr()));
+    }
     return list;
   }
 
@@ -141,6 +146,23 @@
     }
   }
 
+  private final class SendAfterCancellationGen implements Gen {
+
+    @Override
+    public void generate(Commands.Builder builder, int depth) {
+      final String token = new UUID(random.nextLong(), random.nextLong()).toString();
+      final int address = address();
+
+      Command.SendAfter.Builder first =
+          Command.SendAfter.newBuilder().setTarget(address).setCancellationToken(token);
+      Command.CancelSendAfter.Builder second =
+          Command.CancelSendAfter.newBuilder().setTarget(address).setCancellationToken(token);
+
+      builder.addCommand(Command.newBuilder().setSendAfter(first));
+      builder.addCommand(Command.newBuilder().setCancelSendAfter(second));
+    }
+  }
+
   private final class SendGen implements Gen {
 
     @Override
diff --git a/statefun-e2e-tests/statefun-smoke-e2e-embedded/src/main/java/org/apache/flink/statefun/e2e/smoke/embedded/CommandInterpreter.java b/statefun-e2e-tests/statefun-smoke-e2e-embedded/src/main/java/org/apache/flink/statefun/e2e/smoke/embedded/CommandInterpreter.java
index a4da76d..f184929 100644
--- a/statefun-e2e-tests/statefun-smoke-e2e-embedded/src/main/java/org/apache/flink/statefun/e2e/smoke/embedded/CommandInterpreter.java
+++ b/statefun-e2e-tests/statefun-smoke-e2e-embedded/src/main/java/org/apache/flink/statefun/e2e/smoke/embedded/CommandInterpreter.java
@@ -17,7 +17,13 @@
  */
 package org.apache.flink.statefun.e2e.smoke.embedded;
 
-import static org.apache.flink.statefun.e2e.smoke.driver.Types.*;
+import static org.apache.flink.statefun.e2e.smoke.driver.Types.COMMANDS_TYPE;
+import static org.apache.flink.statefun.e2e.smoke.driver.Types.SOURCE_COMMANDS_TYPE;
+import static org.apache.flink.statefun.e2e.smoke.driver.Types.isTypeOf;
+import static org.apache.flink.statefun.e2e.smoke.driver.Types.packCommands;
+import static org.apache.flink.statefun.e2e.smoke.driver.Types.packVerificationResult;
+import static org.apache.flink.statefun.e2e.smoke.driver.Types.unpackCommands;
+import static org.apache.flink.statefun.e2e.smoke.driver.Types.unpackSourceCommand;
 
 import java.time.Duration;
 import java.util.Objects;
@@ -28,6 +34,7 @@
 import org.apache.flink.statefun.e2e.smoke.generated.Commands;
 import org.apache.flink.statefun.e2e.smoke.generated.SourceCommand;
 import org.apache.flink.statefun.e2e.smoke.generated.VerificationResult;
+import org.apache.flink.statefun.sdk.Address;
 import org.apache.flink.statefun.sdk.AsyncOperationResult;
 import org.apache.flink.statefun.sdk.Context;
 import org.apache.flink.statefun.sdk.FunctionType;
@@ -81,6 +88,8 @@
         sendEgress(state, context, cmd.getSendEgress());
       } else if (cmd.hasVerify()) {
         verify(state, context, cmd.getVerify());
+      } else if (cmd.hasCancelSendAfter()) {
+        cancelSendAfter(state, context, cmd.getCancelSendAfter());
       }
     }
   }
@@ -114,7 +123,21 @@
       Command.SendAfter send) {
     FunctionType functionType = Constants.FN_TYPE;
     String id = ids.idOf(send.getTarget());
-    context.sendAfter(sendAfterDelay, functionType, id, packCommands(send.getCommands()));
+    TypedValue subCommands = packCommands(send.getCommands());
+    if (send.getCancellationToken().isEmpty()) {
+      context.sendAfter(sendAfterDelay, functionType, id, subCommands);
+    } else {
+      context.sendAfter(
+          sendAfterDelay, new Address(functionType, id), subCommands, send.getCancellationToken());
+    }
+  }
+
+  private void cancelSendAfter(
+      @SuppressWarnings("unused") PersistedValue<Long> state,
+      Context context,
+      Command.CancelSendAfter cancelSendAfter) {
+    String token = cancelSendAfter.getCancellationToken();
+    context.cancelDelayedMessage(token);
   }
 
   private void send(
diff --git a/statefun-e2e-tests/statefun-smoke-e2e-embedded/src/test/java/org/apache/flink/statefun/e2e/smoke/embedded/EmbeddedSmokeHarnessTest.java b/statefun-e2e-tests/statefun-smoke-e2e-embedded/src/test/java/org/apache/flink/statefun/e2e/smoke/embedded/EmbeddedSmokeHarnessTest.java
index 925ebd1..6760418 100644
--- a/statefun-e2e-tests/statefun-smoke-e2e-embedded/src/test/java/org/apache/flink/statefun/e2e/smoke/embedded/EmbeddedSmokeHarnessTest.java
+++ b/statefun-e2e-tests/statefun-smoke-e2e-embedded/src/test/java/org/apache/flink/statefun/e2e/smoke/embedded/EmbeddedSmokeHarnessTest.java
@@ -61,6 +61,7 @@
     parameters.setVerificationServerHost("localhost");
     parameters.setVerificationServerPort(started.port());
     parameters.setAsyncOpSupported(true);
+    parameters.setDelayCancellationOpSupported(true);
     parameters.asMap().forEach(harness::withGlobalConfiguration);
 
     // run the harness.
diff --git a/statefun-e2e-tests/statefun-smoke-e2e-embedded/src/test/java/org/apache/flink/statefun/e2e/smoke/embedded/SmokeVerificationEmbeddedE2E.java b/statefun-e2e-tests/statefun-smoke-e2e-embedded/src/test/java/org/apache/flink/statefun/e2e/smoke/embedded/SmokeVerificationEmbeddedE2E.java
index cc7a0b4..63f3afb 100644
--- a/statefun-e2e-tests/statefun-smoke-e2e-embedded/src/test/java/org/apache/flink/statefun/e2e/smoke/embedded/SmokeVerificationEmbeddedE2E.java
+++ b/statefun-e2e-tests/statefun-smoke-e2e-embedded/src/test/java/org/apache/flink/statefun/e2e/smoke/embedded/SmokeVerificationEmbeddedE2E.java
@@ -34,6 +34,7 @@
     parameters.setMessageCount(100_000);
     parameters.setMaxFailures(1);
     parameters.setAsyncOpSupported(true);
+    parameters.setDelayCancellationOpSupported(true);
 
     StatefulFunctionsAppContainers.Builder builder =
         StatefulFunctionsAppContainers.builder("flink-statefun-cluster", NUM_WORKERS);