Skip to content

test: add payload offloader runners and integration coverage #682

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zhongkechen wants to merge 1 commit into from
Open

test: add payload offloader runners and integration coverage #682

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions sdk-integration-tests/src/test/java/software/amazon/lambda/durable/MapIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -586,8 +586,9 @@ void testMapAsyncWithInterleavedWork(NestingType nestingType, int events) {
}

@ParameterizedTest
@CsvSource({"FLAT, 2", "NESTED, 12"})
void testMapUnlimitedConcurrencyWithToleratedFailureCount(NestingType nestingType, int events) {
@CsvSource({"FLAT, 2, 2", "NESTED, 11, 12"})
void testMapUnlimitedConcurrencyWithToleratedFailureCount(
NestingType nestingType, int minimumEvents, int maximumEvents) {
var runner = LocalDurableTestRunner.create(String.class, (input, context) -> {
var items = List.of("ok1", "FAIL1", "ok2", "FAIL2", "ok3");
var config = MapConfig.builder()
Expand All @@ -613,7 +614,8 @@ void testMapUnlimitedConcurrencyWithToleratedFailureCount(NestingType nestingTyp

var result = runner.runUntilComplete("test");
assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
assertEquals(events, result.getHistoryEvents().size());
assertTrue(minimumEvents <= result.getHistoryEvents().size());
assertTrue(result.getHistoryEvents().size() <= maximumEvents);
}

@Test
Expand Down
8 changes: 5 additions & 3 deletions ...tegration-tests/src/test/java/software/amazon/lambda/durable/ParallelIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -526,8 +526,9 @@ void testParallelWithWaitInsideBranches_replay(NestingType nestingType, int even
}

@ParameterizedTest
@CsvSource({"FLAT, 2", "NESTED, 12"})
void testParallelUnlimitedConcurrencyWithToleratedFailureCount(NestingType nestingType, int events) {
@CsvSource({"FLAT, 2, 2", "NESTED, 11, 12"})
void testParallelUnlimitedConcurrencyWithToleratedFailureCount(
NestingType nestingType, int minimumEvents, int maximumEvents) {
var runner = LocalDurableTestRunner.create(String.class, (input, context) -> {
var config = ParallelConfig.builder()
.completionConfig(CompletionConfig.toleratedFailureCount(1))
Expand Down Expand Up @@ -555,7 +556,8 @@ void testParallelUnlimitedConcurrencyWithToleratedFailureCount(NestingType nesti

var result = runner.runUntilComplete("test");
assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
assertEquals(events, result.getHistoryEvents().size());
assertTrue(minimumEvents <= result.getHistoryEvents().size());
assertTrue(result.getHistoryEvents().size() <= maximumEvents);
}

@ParameterizedTest
Expand Down
1,354 changes: 1,354 additions & 0 deletions ...n-tests/src/test/java/software/amazon/lambda/durable/PayloadOffloaderIntegrationTest.java
View file
Open in desktop

Large diffs are not rendered by default.

29 changes: 28 additions & 1 deletion sdk-testing/src/main/java/software/amazon/lambda/durable/testing/AsyncExecution.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import java.time.Duration;
import java.time.Instant;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.function.Predicate;
import software.amazon.awssdk.core.SdkBytes;
import software.amazon.awssdk.services.lambda.LambdaClient;
Expand All @@ -14,7 +15,9 @@
import software.amazon.awssdk.services.lambda.model.GetDurableExecutionHistoryRequest;
import software.amazon.awssdk.services.lambda.model.ResourceNotFoundException;
import software.amazon.lambda.durable.TypeToken;
import software.amazon.lambda.durable.execution.PayloadCodec;
import software.amazon.lambda.durable.model.ExecutionStatus;
import software.amazon.lambda.durable.offload.PayloadOffloader;
import software.amazon.lambda.durable.serde.SerDes;
import software.amazon.lambda.durable.testing.cloud.HistoryEventProcessor;

Expand All @@ -27,25 +30,43 @@ public class AsyncExecution<O> {
private final LambdaClient lambdaClient;
private final TypeToken<O> outputType;
private final SerDes serDes;
private final PayloadOffloader payloadOffloader;
private final ExecutorService payloadOffloadExecutorService;
private final Duration pollInterval;
private final Duration timeout;
private final HistoryEventProcessor processor;
private List<Event> currentHistory;
private TestResult<O> currentResult;

/** Creates an execution handle without payload-offloader-aware history decoding. */
public AsyncExecution(
String executionArn,
LambdaClient lambdaClient,
TypeToken<O> outputType,
SerDes serDes,
Duration pollInterval,
Duration timeout) {
this(executionArn, lambdaClient, outputType, serDes, null, null, pollInterval, timeout);
}

/** Creates an execution handle with optional payload-offloader-aware history decoding. */
public AsyncExecution(
String executionArn,
LambdaClient lambdaClient,
TypeToken<O> outputType,
SerDes serDes,
PayloadOffloader payloadOffloader,
ExecutorService payloadOffloadExecutorService,
Duration pollInterval,
Duration timeout) {
this.executionArn = executionArn;
this.lambdaClient = lambdaClient;
this.outputType = outputType;
this.pollInterval = pollInterval;
this.timeout = timeout;
this.serDes = serDes;
this.payloadOffloader = payloadOffloader;
this.payloadOffloadExecutorService = payloadOffloadExecutorService;
this.processor = new HistoryEventProcessor();
}

Expand Down Expand Up @@ -195,7 +216,13 @@ private void refreshHistory() {
.build();
var response = lambdaClient.getDurableExecutionHistory(request);
this.currentHistory = response.events();
this.currentResult = processor.processEvents(currentHistory, outputType, serDes);
this.currentResult = processor.processEvents(
currentHistory,
outputType,
serDes,
new PayloadCodec(payloadOffloadExecutorService),
payloadOffloader,
executionArn);
} catch (ResourceNotFoundException e) {
// Execution doesn't exist yet - this can happen immediately after async invoke
// Leave currentHistory as null, pollUntil will retry
Expand Down
118 changes: 110 additions & 8 deletions sdk-testing/src/main/java/software/amazon/lambda/durable/testing/CloudDurableTestRunner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,12 +4,15 @@

import java.time.Duration;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
import software.amazon.awssdk.core.SdkBytes;
import software.amazon.awssdk.services.lambda.LambdaClient;
import software.amazon.awssdk.services.lambda.model.InvocationType;
import software.amazon.awssdk.services.lambda.model.InvokeRequest;
import software.amazon.lambda.durable.TypeToken;
import software.amazon.lambda.durable.execution.PayloadCodec;
import software.amazon.lambda.durable.offload.PayloadOffloader;
import software.amazon.lambda.durable.serde.JacksonSerDes;
import software.amazon.lambda.durable.serde.SerDes;
import software.amazon.lambda.durable.testing.cloud.HistoryEventProcessor;
Expand All @@ -31,6 +34,8 @@ public class CloudDurableTestRunner<I, O> {
private final Duration timeout;
private final InvocationType invocationType;
private final SerDes serDes;
private final PayloadOffloader payloadOffloader;
private final ExecutorService payloadOffloadExecutorService;
// Store last execution result for operation inspection
private TestResult<O> lastResult;

Expand All @@ -42,7 +47,9 @@ private CloudDurableTestRunner(
Duration pollInterval,
Duration timeout,
InvocationType invocationType,
SerDes serDes) {
SerDes serDes,
PayloadOffloader payloadOffloader,
ExecutorService payloadOffloadExecutorService) {
this.functionArn = functionArn;
this.inputType = inputType;
this.outputType = outputType;
Expand All @@ -52,6 +59,8 @@ private CloudDurableTestRunner(
this.timeout = timeout;
this.invocationType = invocationType;
this.serDes = Objects.requireNonNullElseGet(serDes, JacksonSerDes::new);
this.payloadOffloader = payloadOffloader;
this.payloadOffloadExecutorService = payloadOffloadExecutorService;
}

private static LambdaClient createDefaultLambdaClient() {
Expand All @@ -77,6 +86,8 @@ public static <I, O> CloudDurableTestRunner<I, O> create(
Duration.ofSeconds(2),
Duration.ofSeconds(300),
InvocationType.REQUEST_RESPONSE,
null,
null,
null);
}

Expand All @@ -97,36 +108,113 @@ public static <I, O> CloudDurableTestRunner<I, O> create(
Duration.ofSeconds(2),
Duration.ofSeconds(300),
InvocationType.REQUEST_RESPONSE,
null,
null,
null);
}

/** Returns a new runner with the specified lambda client. */
public CloudDurableTestRunner<I, O> withLambdaClient(LambdaClient lambdaClient) {
return new CloudDurableTestRunner<>(
functionArn, inputType, outputType, lambdaClient, pollInterval, timeout, invocationType, serDes);
functionArn,
inputType,
outputType,
lambdaClient,
pollInterval,
timeout,
invocationType,
serDes,
payloadOffloader,
payloadOffloadExecutorService);
}

/** Returns a new runner with the specified poll interval between history checks. */
public CloudDurableTestRunner<I, O> withPollInterval(Duration interval) {
return new CloudDurableTestRunner<>(
functionArn, inputType, outputType, lambdaClient, interval, timeout, invocationType, serDes);
functionArn,
inputType,
outputType,
lambdaClient,
interval,
timeout,
invocationType,
serDes,
payloadOffloader,
payloadOffloadExecutorService);
}

/** Returns a new runner with the specified maximum wait time for execution completion. */
public CloudDurableTestRunner<I, O> withTimeout(Duration timeout) {
return new CloudDurableTestRunner<>(
functionArn, inputType, outputType, lambdaClient, pollInterval, timeout, invocationType, serDes);
functionArn,
inputType,
outputType,
lambdaClient,
pollInterval,
timeout,
invocationType,
serDes,
payloadOffloader,
payloadOffloadExecutorService);
}

/** Returns a new runner with the specified Lambda invocation type. */
public CloudDurableTestRunner<I, O> withInvocationType(InvocationType type) {
return new CloudDurableTestRunner<>(
functionArn, inputType, outputType, lambdaClient, pollInterval, timeout, type, serDes);
functionArn,
inputType,
outputType,
lambdaClient,
pollInterval,
timeout,
type,
serDes,
payloadOffloader,
payloadOffloadExecutorService);
}

public CloudDurableTestRunner<I, O> withSerDes(SerDes serDes) {
return new CloudDurableTestRunner<>(
functionArn, inputType, outputType, lambdaClient, pollInterval, timeout, invocationType, serDes);
functionArn,
inputType,
outputType,
lambdaClient,
pollInterval,
timeout,
invocationType,
serDes,
payloadOffloader,
payloadOffloadExecutorService);
}

/** Returns a new runner that resolves payloads with the supplied offloader. */
public CloudDurableTestRunner<I, O> withPayloadOffloader(PayloadOffloader payloadOffloader) {
return new CloudDurableTestRunner<>(
functionArn,
inputType,
outputType,
lambdaClient,
pollInterval,
timeout,
invocationType,
serDes,
Objects.requireNonNull(payloadOffloader, "payloadOffloader cannot be null"),
payloadOffloadExecutorService);
}

/** Returns a new runner with a dedicated payload I/O executor. */
public CloudDurableTestRunner<I, O> withPayloadOffloadExecutorService(ExecutorService executorService) {
return new CloudDurableTestRunner<>(
functionArn,
inputType,
outputType,
lambdaClient,
pollInterval,
timeout,
invocationType,
serDes,
payloadOffloader,
Objects.requireNonNull(executorService, "executorService cannot be null"));
}

/** Invokes the Lambda function, polls execution history until completion, and returns the result. */
Expand Down Expand Up @@ -161,7 +249,13 @@ public TestResult<O> run(I input) {

// Process events into TestResult
var processor = new HistoryEventProcessor();
var result = processor.processEvents(events, outputType, serDes);
var result = processor.processEvents(
events,
outputType,
serDes,
new PayloadCodec(payloadOffloadExecutorService),
payloadOffloader,
executionArn);
this.lastResult = result;
return result;
} catch (Exception e) {
Expand Down Expand Up @@ -200,7 +294,15 @@ public AsyncExecution<O> startAsync(I input) {
// This prevents immediate polling from failing with "execution does not exist"
Thread.sleep(100);

return new AsyncExecution<>(executionArn, lambdaClient, outputType, serDes, pollInterval, timeout);
return new AsyncExecution<>(
executionArn,
lambdaClient,
outputType,
serDes,
payloadOffloader,
payloadOffloadExecutorService,
pollInterval,
timeout);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException("Interrupted while starting async execution", e);
Expand Down
Loading
Loading