fix anyOf with empty map

Author: zhongkechen

sdk-integration-tests/src/test/java/software/amazon/lambda/durable/MapIntegrationTest.java

@@ -928,4 +928,20 @@ void testMultipleMapAsyncInParallel() {
         assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
         assertEquals("2,4,6|A,B|olleh,dlrow,oof,rab", result.getResult(String.class));
     }
+
+    @Test
+    void testMapWithEmptyItems() {
+        var runner = LocalDurableTestRunner.create(String.class, (input, context) -> {
+            List<String> items = List.of();
+            var result = context.map("empty-map", items, String.class, (item, index, ctx) -> item);
+
+            assertTrue(result.allSucceeded());
+            assertEquals(0, result.size());
+            assertTrue(result.results().isEmpty());
+            return "done";
+        });
+
+        var result = runner.runUntilComplete("test");
+        assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
+    }
 }

sdk/src/main/java/software/amazon/lambda/durable/context/DurableContextImpl.java

@@ -43,7 +43,6 @@
 import software.amazon.lambda.durable.operation.StepOperation;
 import software.amazon.lambda.durable.operation.WaitForConditionOperation;
 import software.amazon.lambda.durable.operation.WaitOperation;
-import software.amazon.lambda.durable.util.CompletedDurableFuture;
 import software.amazon.lambda.durable.util.ParameterValidator;
 
 /**
@@ -256,11 +255,6 @@ public <I, O> DurableFuture<MapResult<O>> mapAsync(
             config = config.toBuilder().serDes(getDurableConfig().getSerDes()).build();
         }
 
-        // Short-circuit for empty collections — no checkpoint overhead
-        if (items.isEmpty()) {
-            return new CompletedDurableFuture<>(MapResult.empty());
-        }
-
         // Convert to List for deterministic index-based access
         var itemList = List.copyOf(items);
         var operationId = nextOperationId();

sdk/src/main/java/software/amazon/lambda/durable/operation/MapOperation.java

@@ -111,6 +111,9 @@ private static Integer getToleratedFailureCount(CompletionConfig completionConfi
 
     @Override
     protected void start() {
+        if (items.isEmpty()) {
+            return;
+        }
         sendOperationUpdateAsync(OperationUpdate.builder()
                 .action(OperationAction.START)
                 .subType(getSubType().getValue()));
@@ -120,6 +123,10 @@ protected void start() {
 
     @Override
     protected void replay(Operation existing) {
+        if (items.isEmpty()) {
+            markAlreadyCompleted();
+            return;
+        }
         switch (existing.status()) {
             case SUCCEEDED -> {
                 if (existing.contextDetails() != null
@@ -194,6 +201,9 @@ protected void handleCompletion(ConcurrencyCompletionStatus concurrencyCompletio
 
     @Override
     public MapResult<O> get() {
+        if (items.isEmpty()) {
+            return MapResult.empty();
+        }
         if (replayFromPayload) {
             // Small result replay: deserialize MapResult directly from checkpoint payload
             var op = waitForOperationCompletion();