diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index b4788aa64..52669b5de 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -118,7 +118,7 @@ jobs: GOPROXY: https://proxy.golang.org JDK_VER: ${{ matrix.java }} DAPR_CLI_VER: 1.17.0 - DAPR_RUNTIME_VER: 1.18.0-rc.3 + DAPR_RUNTIME_VER: 1.18.0-rc.4 DAPR_INSTALL_URL: https://raw.githubusercontent.com/dapr/cli/v1.17.0/install/install.sh DAPR_CLI_REF: DAPR_REF: diff --git a/.github/workflows/validate.yml b/.github/workflows/validate.yml index e9a8d88cc..bb273a797 100644 --- a/.github/workflows/validate.yml +++ b/.github/workflows/validate.yml @@ -38,7 +38,7 @@ jobs: GOPROXY: https://proxy.golang.org JDK_VER: ${{ matrix.java }} DAPR_CLI_VER: 1.17.0 - DAPR_RUNTIME_VER: 1.18.0-rc.3 + DAPR_RUNTIME_VER: 1.18.0-rc.4 DAPR_INSTALL_URL: https://raw.githubusercontent.com/dapr/cli/v1.17.0/install/install.sh DAPR_CLI_REF: DAPR_REF: diff --git a/pom.xml b/pom.xml index 9b1eea449..b67bab16e 100644 --- a/pom.xml +++ b/pom.xml @@ -16,7 +16,7 @@ UTF-8 1.79.0 3.25.5 - https://raw.githubusercontent.com/dapr/dapr/v1.18.0-rc.3/dapr/proto + https://raw.githubusercontent.com/dapr/dapr/v1.18.0-rc.4/dapr/proto https://raw.githubusercontent.com/dapr/durabletask-protobuf/main/protos 1.18.0-rc-2 1.7.1 diff --git a/sdk/src/main/java/io/dapr/client/DaprClientImpl.java b/sdk/src/main/java/io/dapr/client/DaprClientImpl.java index a6b6bbbc3..de91813c5 100644 --- a/sdk/src/main/java/io/dapr/client/DaprClientImpl.java +++ b/sdk/src/main/java/io/dapr/client/DaprClientImpl.java @@ -1517,13 +1517,23 @@ public Mono scheduleJob(ScheduleJobRequest scheduleJobRequest) { jobBuilder.setFailurePolicy(FailurePolicyUtils.getJobFailurePolicy(scheduleJobRequest.getFailurePolicy())); } + DaprJobsProtos.ScheduleJobRequest request = DaprJobsProtos.ScheduleJobRequest.newBuilder() + .setOverwrite(scheduleJobRequest.getOverwrite()) + .setJob(jobBuilder.build()) + .build(); + Mono scheduleJobResponseMono = - Mono.deferContextual(context -> this.createMono( - it -> intercept(context, asyncStub) - .scheduleJobAlpha1(DaprJobsProtos.ScheduleJobRequest.newBuilder() - .setOverwrite(scheduleJobRequest.getOverwrite()) - .setJob(jobBuilder.build()).build(), it) - ) + Mono.deferContextual(context -> + this.createMono( + it -> intercept(context, asyncStub).scheduleJob(request, it) + ).onErrorResume(throwable -> { + // The stable ScheduleJob RPC is unavailable on pre-1.18 sidecars, fall back to Alpha1. + if (isUnimplemented(throwable)) { + return this.createMono( + it -> intercept(context, asyncStub).scheduleJobAlpha1(request, it)); + } + return Mono.error(throwable); + }) ); return scheduleJobResponseMono.then(); @@ -1539,12 +1549,22 @@ public Mono getJob(GetJobRequest getJobRequest) { try { validateGetJobRequest(getJobRequest); + DaprJobsProtos.GetJobRequest request = DaprJobsProtos.GetJobRequest.newBuilder() + .setName(getJobRequest.getName()) + .build(); + Mono getJobResponseMono = - Mono.deferContextual(context -> this.createMono( - it -> intercept(context, asyncStub) - .getJobAlpha1(DaprJobsProtos.GetJobRequest.newBuilder() - .setName(getJobRequest.getName()).build(), it) - ) + Mono.deferContextual(context -> + this.createMono( + it -> intercept(context, asyncStub).getJob(request, it) + ).onErrorResume(throwable -> { + // The stable GetJob RPC is unavailable on pre-1.18 sidecars, fall back to Alpha1. + if (isUnimplemented(throwable)) { + return this.createMono( + it -> intercept(context, asyncStub).getJobAlpha1(request, it)); + } + return Mono.error(throwable); + }) ); return getJobResponseMono.map(response -> { @@ -1602,12 +1622,22 @@ public Mono deleteJob(DeleteJobRequest deleteJobRequest) { try { validateDeleteJobRequest(deleteJobRequest); + DaprJobsProtos.DeleteJobRequest request = DaprJobsProtos.DeleteJobRequest.newBuilder() + .setName(deleteJobRequest.getName()) + .build(); + Mono deleteJobResponseMono = - Mono.deferContextual(context -> this.createMono( - it -> intercept(context, asyncStub) - .deleteJobAlpha1(DaprJobsProtos.DeleteJobRequest.newBuilder() - .setName(deleteJobRequest.getName()).build(), it) - ) + Mono.deferContextual(context -> + this.createMono( + it -> intercept(context, asyncStub).deleteJob(request, it) + ).onErrorResume(throwable -> { + // The stable DeleteJob RPC is unavailable on pre-1.18 sidecars, fall back to Alpha1. + if (isUnimplemented(throwable)) { + return this.createMono( + it -> intercept(context, asyncStub).deleteJobAlpha1(request, it)); + } + return Mono.error(throwable); + }) ); return deleteJobResponseMono.then(); @@ -1616,6 +1646,16 @@ public Mono deleteJob(DeleteJobRequest deleteJobRequest) { } } + /** + * Returns true when a failed RPC indicates the sidecar has not implemented the stable RPC, + * so the caller should retry against the deprecated Alpha1 RPC. The gRPC status is wrapped in a + * {@link DaprException} (and an {@link java.util.concurrent.ExecutionException}), so we let gRPC + * walk the cause chain to recover the original status code. + */ + private static boolean isUnimplemented(Throwable throwable) { + return io.grpc.Status.fromThrowable(throwable).getCode() == io.grpc.Status.Code.UNIMPLEMENTED; + } + private void validateScheduleJobRequest(ScheduleJobRequest scheduleJobRequest) { if (scheduleJobRequest == null) { throw new IllegalArgumentException("scheduleJobRequest cannot be null"); diff --git a/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java b/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java index ea31a4880..6af09aa19 100644 --- a/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java +++ b/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java @@ -2302,14 +2302,14 @@ public void scheduleJobShouldSucceedWhenAllFieldsArePresentInRequest() { StreamObserver observer = invocation.getArgument(1); observer.onCompleted(); // Simulate successful response return null; - }).when(daprStub).scheduleJobAlpha1(any(DaprJobsProtos.ScheduleJobRequest.class), any()); + }).when(daprStub).scheduleJob(any(DaprJobsProtos.ScheduleJobRequest.class), any()); assertDoesNotThrow(() -> client.scheduleJob(expectedScheduleJobRequest).block()); ArgumentCaptor captor = ArgumentCaptor.forClass(DaprJobsProtos.ScheduleJobRequest.class); - verify(daprStub, times(1)).scheduleJobAlpha1(captor.capture(), Mockito.any()); + verify(daprStub, times(1)).scheduleJob(captor.capture(), Mockito.any()); DaprJobsProtos.ScheduleJobRequest actualScheduleJobReq = captor.getValue(); assertEquals("testJob", actualScheduleJobReq.getJob().getName()); @@ -2321,6 +2321,31 @@ public void scheduleJobShouldSucceedWhenAllFieldsArePresentInRequest() { assertEquals(iso8601Formatter.format(expectedScheduleJobRequest.getDueTime()), actualScheduleJobReq.getJob().getDueTime()); } + @Test + public void scheduleJobShouldFallBackToAlpha1WhenStableRpcIsUnimplemented() { + ScheduleJobRequest scheduleJobRequest = new ScheduleJobRequest("testJob", + JobSchedule.fromString("*/5 * * * *")); + + // The stable ScheduleJob RPC is not implemented on the (older) sidecar... + doAnswer(invocation -> { + StreamObserver observer = invocation.getArgument(1); + observer.onError(newStatusRuntimeException("UNIMPLEMENTED", "stable ScheduleJob not implemented")); + return null; + }).when(daprStub).scheduleJob(any(DaprJobsProtos.ScheduleJobRequest.class), any()); + + // ...so the client falls back to the deprecated Alpha1 RPC. + doAnswer(invocation -> { + StreamObserver observer = invocation.getArgument(1); + observer.onCompleted(); + return null; + }).when(daprStub).scheduleJobAlpha1(any(DaprJobsProtos.ScheduleJobRequest.class), any()); + + assertDoesNotThrow(() -> client.scheduleJob(scheduleJobRequest).block()); + + verify(daprStub, times(1)).scheduleJob(any(DaprJobsProtos.ScheduleJobRequest.class), Mockito.any()); + verify(daprStub, times(1)).scheduleJobAlpha1(any(DaprJobsProtos.ScheduleJobRequest.class), Mockito.any()); + } + @Test public void scheduleJobShouldSucceedWhenRequiredFieldsNameAndDueTimeArePresentInRequest() { DateTimeFormatter iso8601Formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'") @@ -2330,7 +2355,7 @@ public void scheduleJobShouldSucceedWhenRequiredFieldsNameAndDueTimeArePresentIn StreamObserver observer = invocation.getArgument(1); observer.onCompleted(); // Simulate successful response return null; - }).when(daprStub).scheduleJobAlpha1(any(DaprJobsProtos.ScheduleJobRequest.class), any()); + }).when(daprStub).scheduleJob(any(DaprJobsProtos.ScheduleJobRequest.class), any()); ScheduleJobRequest expectedScheduleJobRequest = new ScheduleJobRequest("testJob", Instant.now().plus(10, ChronoUnit.MINUTES)); @@ -2339,7 +2364,7 @@ public void scheduleJobShouldSucceedWhenRequiredFieldsNameAndDueTimeArePresentIn ArgumentCaptor captor = ArgumentCaptor.forClass(DaprJobsProtos.ScheduleJobRequest.class); - verify(daprStub, times(1)).scheduleJobAlpha1(captor.capture(), Mockito.any()); + verify(daprStub, times(1)).scheduleJob(captor.capture(), Mockito.any()); DaprJobsProtos.ScheduleJobRequest actualScheduleJobRequest = captor.getValue(); DaprJobsProtos.Job job = actualScheduleJobRequest.getJob(); assertEquals("testJob", job.getName()); @@ -2360,7 +2385,7 @@ public void scheduleJobShouldSucceedWhenRequiredFieldsNameAndScheduleArePresentI StreamObserver observer = invocation.getArgument(1); observer.onCompleted(); // Simulate successful response return null; - }).when(daprStub).scheduleJobAlpha1(any(DaprJobsProtos.ScheduleJobRequest.class), any()); + }).when(daprStub).scheduleJob(any(DaprJobsProtos.ScheduleJobRequest.class), any()); ScheduleJobRequest expectedScheduleJobRequest = new ScheduleJobRequest("testJob", JobSchedule.fromString("* * * * * *")); @@ -2369,7 +2394,7 @@ public void scheduleJobShouldSucceedWhenRequiredFieldsNameAndScheduleArePresentI ArgumentCaptor captor = ArgumentCaptor.forClass(DaprJobsProtos.ScheduleJobRequest.class); - verify(daprStub, times(1)).scheduleJobAlpha1(captor.capture(), Mockito.any()); + verify(daprStub, times(1)).scheduleJob(captor.capture(), Mockito.any()); DaprJobsProtos.ScheduleJobRequest actualScheduleJobRequest = captor.getValue(); DaprJobsProtos.Job job = actualScheduleJobRequest.getJob(); assertEquals("testJob", job.getName()); @@ -2385,7 +2410,7 @@ public void scheduleJobShouldHavePolicyWhenPolicyIsSet() { StreamObserver observer = invocation.getArgument(1); observer.onCompleted(); // Simulate successful response return null; - }).when(daprStub).scheduleJobAlpha1(any(DaprJobsProtos.ScheduleJobRequest.class), any()); + }).when(daprStub).scheduleJob(any(DaprJobsProtos.ScheduleJobRequest.class), any()); ScheduleJobRequest expectedScheduleJobRequest = new ScheduleJobRequest("testJob", JobSchedule.fromString("* * * * * *")) @@ -2396,7 +2421,7 @@ public void scheduleJobShouldHavePolicyWhenPolicyIsSet() { ArgumentCaptor captor = ArgumentCaptor.forClass(DaprJobsProtos.ScheduleJobRequest.class); - verify(daprStub, times(1)).scheduleJobAlpha1(captor.capture(), Mockito.any()); + verify(daprStub, times(1)).scheduleJob(captor.capture(), Mockito.any()); DaprJobsProtos.ScheduleJobRequest actualScheduleJobRequest = captor.getValue(); DaprJobsProtos.Job job = actualScheduleJobRequest.getJob(); assertEquals("testJob", job.getName()); @@ -2440,7 +2465,7 @@ public void scheduleJobShouldHaveConstantPolicyWithMaxRetriesWhenConstantPolicyI StreamObserver observer = invocation.getArgument(1); observer.onCompleted(); // Simulate successful response return null; - }).when(daprStub).scheduleJobAlpha1(any(DaprJobsProtos.ScheduleJobRequest.class), any()); + }).when(daprStub).scheduleJob(any(DaprJobsProtos.ScheduleJobRequest.class), any()); ScheduleJobRequest expectedScheduleJobRequest = new ScheduleJobRequest("testJob", JobSchedule.fromString("* * * * * *")) @@ -2451,7 +2476,7 @@ public void scheduleJobShouldHaveConstantPolicyWithMaxRetriesWhenConstantPolicyI ArgumentCaptor captor = ArgumentCaptor.forClass(DaprJobsProtos.ScheduleJobRequest.class); - verify(daprStub, times(1)).scheduleJobAlpha1(captor.capture(), Mockito.any()); + verify(daprStub, times(1)).scheduleJob(captor.capture(), Mockito.any()); DaprJobsProtos.ScheduleJobRequest actualScheduleJobRequest = captor.getValue(); DaprJobsProtos.Job job = actualScheduleJobRequest.getJob(); assertEquals("testJob", job.getName()); @@ -2469,7 +2494,7 @@ public void scheduleJobShouldHaveConstantPolicyWithIntervalWhenConstantPolicyIsS StreamObserver observer = invocation.getArgument(1); observer.onCompleted(); // Simulate successful response return null; - }).when(daprStub).scheduleJobAlpha1(any(DaprJobsProtos.ScheduleJobRequest.class), any()); + }).when(daprStub).scheduleJob(any(DaprJobsProtos.ScheduleJobRequest.class), any()); ScheduleJobRequest expectedScheduleJobRequest = new ScheduleJobRequest("testJob", JobSchedule.fromString("* * * * * *")) @@ -2480,7 +2505,7 @@ public void scheduleJobShouldHaveConstantPolicyWithIntervalWhenConstantPolicyIsS ArgumentCaptor captor = ArgumentCaptor.forClass(DaprJobsProtos.ScheduleJobRequest.class); - verify(daprStub, times(1)).scheduleJobAlpha1(captor.capture(), Mockito.any()); + verify(daprStub, times(1)).scheduleJob(captor.capture(), Mockito.any()); DaprJobsProtos.ScheduleJobRequest actualScheduleJobRequest = captor.getValue(); DaprJobsProtos.Job job = actualScheduleJobRequest.getJob(); assertEquals("testJob", job.getName()); @@ -2499,7 +2524,7 @@ public void scheduleJobShouldHaveBothRetiresAndIntervalWhenConstantPolicyIsSetWi StreamObserver observer = invocation.getArgument(1); observer.onCompleted(); // Simulate successful response return null; - }).when(daprStub).scheduleJobAlpha1(any(DaprJobsProtos.ScheduleJobRequest.class), any()); + }).when(daprStub).scheduleJob(any(DaprJobsProtos.ScheduleJobRequest.class), any()); ScheduleJobRequest expectedScheduleJobRequest = new ScheduleJobRequest("testJob", JobSchedule.fromString("* * * * * *")) @@ -2511,7 +2536,7 @@ public void scheduleJobShouldHaveBothRetiresAndIntervalWhenConstantPolicyIsSetWi ArgumentCaptor captor = ArgumentCaptor.forClass(DaprJobsProtos.ScheduleJobRequest.class); - verify(daprStub, times(1)).scheduleJobAlpha1(captor.capture(), Mockito.any()); + verify(daprStub, times(1)).scheduleJob(captor.capture(), Mockito.any()); DaprJobsProtos.ScheduleJobRequest actualScheduleJobRequest = captor.getValue(); DaprJobsProtos.Job job = actualScheduleJobRequest.getJob(); assertEquals("testJob", job.getName()); @@ -2539,7 +2564,7 @@ public void scheduleJobShouldThrowWhenNameAlreadyExists() { observer.onError(newStatusRuntimeException("ALREADY_EXISTS", "Job with name 'testJob' already exists")); } return null; - }).when(daprStub).scheduleJobAlpha1(any(DaprJobsProtos.ScheduleJobRequest.class), any()); + }).when(daprStub).scheduleJob(any(DaprJobsProtos.ScheduleJobRequest.class), any()); // First call should succeed ScheduleJobRequest firstRequest = new ScheduleJobRequest("testJob", Instant.now()); @@ -2548,7 +2573,7 @@ public void scheduleJobShouldThrowWhenNameAlreadyExists() { ArgumentCaptor captor = ArgumentCaptor.forClass(DaprJobsProtos.ScheduleJobRequest.class); - verify(daprStub, times(1)).scheduleJobAlpha1(captor.capture(), Mockito.any()); + verify(daprStub, times(1)).scheduleJob(captor.capture(), Mockito.any()); DaprJobsProtos.ScheduleJobRequest actualScheduleJobRequest = captor.getValue(); DaprJobsProtos.Job job = actualScheduleJobRequest.getJob(); assertEquals("testJob", job.getName()); @@ -2572,7 +2597,7 @@ public void scheduleJobShouldSucceedWhenNameAlreadyExistsWithOverwrite() { StreamObserver observer = invocation.getArgument(1); observer.onCompleted(); // Simulate successful response for both calls return null; - }).when(daprStub).scheduleJobAlpha1(any(DaprJobsProtos.ScheduleJobRequest.class), any()); + }).when(daprStub).scheduleJob(any(DaprJobsProtos.ScheduleJobRequest.class), any()); // First call should succeed ScheduleJobRequest firstRequest = new ScheduleJobRequest("testJob", Instant.now()); @@ -2586,7 +2611,7 @@ public void scheduleJobShouldSucceedWhenNameAlreadyExistsWithOverwrite() { // Verify that both calls were made successfully ArgumentCaptor captor = ArgumentCaptor.forClass(DaprJobsProtos.ScheduleJobRequest.class); - verify(daprStub, times(2)).scheduleJobAlpha1(captor.capture(), any()); + verify(daprStub, times(2)).scheduleJob(captor.capture(), any()); // Verify the first call doesn't have overwrite set DaprJobsProtos.ScheduleJobRequest firstActualRequest = captor.getAllValues().get(0); @@ -2622,7 +2647,7 @@ public void getJobShouldReturnResponseWhenAllFieldsArePresentInRequest() { .build()); observer.onCompleted(); return null; - }).when(daprStub).getJobAlpha1(any(DaprJobsProtos.GetJobRequest.class), any()); + }).when(daprStub).getJob(any(DaprJobsProtos.GetJobRequest.class), any()); Mono resultMono = client.getJob(getJobRequest); @@ -2652,7 +2677,7 @@ public void getJobShouldReturnResponseWithScheduleSetWhenResponseHasSchedule() { .build()); observer.onCompleted(); return null; - }).when(daprStub).getJobAlpha1(any(DaprJobsProtos.GetJobRequest.class), any()); + }).when(daprStub).getJob(any(DaprJobsProtos.GetJobRequest.class), any()); Mono resultMono = client.getJob(getJobRequest); @@ -2683,7 +2708,7 @@ public void getJobShouldReturnResponseWithDueTimeSetWhenResponseHasDueTime() { .build()); observer.onCompleted(); return null; - }).when(daprStub).getJobAlpha1(any(DaprJobsProtos.GetJobRequest.class), any()); + }).when(daprStub).getJob(any(DaprJobsProtos.GetJobRequest.class), any()); Mono resultMono = client.getJob(getJobRequest); @@ -2716,7 +2741,7 @@ public void getJobShouldReturnResponseWithDropFailurePolicySet() { .build()); observer.onCompleted(); return null; - }).when(daprStub).getJobAlpha1(any(DaprJobsProtos.GetJobRequest.class), any()); + }).when(daprStub).getJob(any(DaprJobsProtos.GetJobRequest.class), any()); Mono resultMono = client.getJob(getJobRequest); @@ -2751,7 +2776,7 @@ public void getJobShouldReturnResponseWithConstantFailurePolicyAndMaxRetriesSet( .build()); observer.onCompleted(); return null; - }).when(daprStub).getJobAlpha1(any(DaprJobsProtos.GetJobRequest.class), any()); + }).when(daprStub).getJob(any(DaprJobsProtos.GetJobRequest.class), any()); Mono resultMono = client.getJob(getJobRequest); @@ -2788,7 +2813,7 @@ public void getJobShouldReturnResponseWithConstantFailurePolicyAndIntervalSet() .build()); observer.onCompleted(); return null; - }).when(daprStub).getJobAlpha1(any(DaprJobsProtos.GetJobRequest.class), any()); + }).when(daprStub).getJob(any(DaprJobsProtos.GetJobRequest.class), any()); Mono resultMono = client.getJob(getJobRequest); @@ -2826,7 +2851,7 @@ public void getJobShouldReturnResponseWithConstantFailurePolicyIntervalAndMaxRet .build()); observer.onCompleted(); return null; - }).when(daprStub).getJobAlpha1(any(DaprJobsProtos.GetJobRequest.class), any()); + }).when(daprStub).getJob(any(DaprJobsProtos.GetJobRequest.class), any()); Mono resultMono = client.getJob(getJobRequest); @@ -2852,7 +2877,7 @@ public void deleteJobShouldSucceedWhenValidRequest() { StreamObserver observer = invocation.getArgument(1); observer.onCompleted(); // Simulate successful response return null; - }).when(daprStub).deleteJobAlpha1(any(DaprJobsProtos.DeleteJobRequest.class), any()); + }).when(daprStub).deleteJob(any(DaprJobsProtos.DeleteJobRequest.class), any()); Mono resultMono = client.deleteJob(deleteJobRequest); diff --git a/testcontainers-dapr/src/main/java/io/dapr/testcontainers/DaprContainerConstants.java b/testcontainers-dapr/src/main/java/io/dapr/testcontainers/DaprContainerConstants.java index 6861510e7..1922b5dd0 100644 --- a/testcontainers-dapr/src/main/java/io/dapr/testcontainers/DaprContainerConstants.java +++ b/testcontainers-dapr/src/main/java/io/dapr/testcontainers/DaprContainerConstants.java @@ -14,7 +14,7 @@ package io.dapr.testcontainers; public interface DaprContainerConstants { - String DAPR_VERSION = "1.18.0-rc.3"; + String DAPR_VERSION = "1.18.0-rc.4"; String DAPR_WORKFLOWS_DASHBOARD_VERSION = "0.0.1"; String DAPR_RUNTIME_IMAGE_TAG = "daprio/daprd:" + DAPR_VERSION; String DAPR_PLACEMENT_IMAGE_TAG = "daprio/placement:" + DAPR_VERSION;