From 24884ea18d0a5e18c34ee77fbbd1e12b454bf726 Mon Sep 17 00:00:00 2001 From: Sky Ao Date: Thu, 3 Dec 2020 16:28:33 +0800 Subject: [PATCH 1/6] add metadata in get bulk state request --- .../java/io/dapr/client/DaprClientGrpc.java | 3 + .../dapr/client/domain/GetStatesRequest.java | 11 + .../domain/GetStatesRequestBuilder.java | 9 + .../io/dapr/client/DaprClientGrpcTest.java | 490 +++++++++--------- 4 files changed, 270 insertions(+), 243 deletions(-) diff --git a/sdk/src/main/java/io/dapr/client/DaprClientGrpc.java b/sdk/src/main/java/io/dapr/client/DaprClientGrpc.java index af13a368f3..959f149205 100644 --- a/sdk/src/main/java/io/dapr/client/DaprClientGrpc.java +++ b/sdk/src/main/java/io/dapr/client/DaprClientGrpc.java @@ -290,6 +290,9 @@ public Mono>>> getStates(GetStatesRequest request, Ty .setStoreName(stateStoreName) .addAllKeys(keys) .setParallelism(parallelism); + if (request.getMetadata() != null) { + builder.putAllMetadata(request.getMetadata()); + } DaprProtos.GetBulkStateRequest envelope = builder.build(); return Mono.fromCallable(wrap(context, () -> { diff --git a/sdk/src/main/java/io/dapr/client/domain/GetStatesRequest.java b/sdk/src/main/java/io/dapr/client/domain/GetStatesRequest.java index e7a2eec2cc..714f534dc8 100644 --- a/sdk/src/main/java/io/dapr/client/domain/GetStatesRequest.java +++ b/sdk/src/main/java/io/dapr/client/domain/GetStatesRequest.java @@ -8,6 +8,7 @@ import io.opentelemetry.context.Context; import java.util.List; +import java.util.Map; /** * A request to get bulk state by keys. @@ -18,6 +19,8 @@ public class GetStatesRequest { private List keys; + private Map metadata; + private int parallelism; private Context context; @@ -53,4 +56,12 @@ public Context getContext() { void setContext(Context context) { this.context = context; } + + public Map getMetadata() { + return metadata; + } + + void setMetadata(Map metadata) { + this.metadata = metadata; + } } diff --git a/sdk/src/main/java/io/dapr/client/domain/GetStatesRequestBuilder.java b/sdk/src/main/java/io/dapr/client/domain/GetStatesRequestBuilder.java index 19479523f3..ddd097e93b 100644 --- a/sdk/src/main/java/io/dapr/client/domain/GetStatesRequestBuilder.java +++ b/sdk/src/main/java/io/dapr/client/domain/GetStatesRequestBuilder.java @@ -10,6 +10,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.Map; /** * Builds a request to request states. @@ -20,6 +21,8 @@ public class GetStatesRequestBuilder { private final List keys; + private Map metadata; + private int parallelism = 1; private Context context; @@ -34,6 +37,11 @@ public GetStatesRequestBuilder(String stateStoreName, String... keys) { this.keys = Collections.unmodifiableList(Arrays.asList(keys)); } + public GetStatesRequestBuilder withMetadata(Map metadata) { + this.metadata = metadata == null ? null : Collections.unmodifiableMap(metadata); + return this; + } + public GetStatesRequestBuilder withParallelism(int parallelism) { this.parallelism = parallelism; return this; @@ -52,6 +60,7 @@ public GetStatesRequest build() { GetStatesRequest request = new GetStatesRequest(); request.setStateStoreName(this.stateStoreName); request.setKeys(this.keys); + request.setMetadata(this.metadata); request.setParallelism(this.parallelism); request.setContext(this.context); return request; diff --git a/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java b/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java index cae8998a0a..b2e9c68064 100644 --- a/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java +++ b/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java @@ -94,7 +94,7 @@ public void tearDown() throws IOException { @Test(expected = RuntimeException.class) public void publishEventExceptionThrownTest() { when(client.publishEvent(any(DaprProtos.PublishEventRequest.class))) - .thenThrow(RuntimeException.class); + .thenThrow(RuntimeException.class); Mono result = adapter.publishEvent("pubsubname","topic", "object"); result.block(); } @@ -106,7 +106,7 @@ public void publishEventCallbackExceptionThrownTest() { MockCallback callback = new MockCallback<>(ex); addCallback(settableFuture, callback, directExecutor()); when(client.publishEvent(any(DaprProtos.PublishEventRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.publishEvent("pubsubname","topic", "object"); settableFuture.setException(ex); result.block(); @@ -118,7 +118,7 @@ public void publishEventSerializeException() throws IOException { adapter = new DaprClientGrpc(closeable, client, mockSerializer, new DefaultObjectSerializer()); SettableFuture settableFuture = SettableFuture.create(); when(client.publishEvent(any(DaprProtos.PublishEventRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); when(mockSerializer.serialize(any())).thenThrow(IOException.class); Mono result = adapter.publishEvent("pubsubname","topic", "{invalid-json"); result.block(); @@ -130,7 +130,7 @@ public void publishEventTest() { MockCallback callback = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFuture, callback, directExecutor()); when(client.publishEvent(any(DaprProtos.PublishEventRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.publishEvent("pubsubname","topic", "object"); settableFuture.set(Empty.newBuilder().build()); result.block(); @@ -143,10 +143,10 @@ public void publishEventNoHotMono() { MockCallback callback = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFuture, callback, directExecutor()); when(client.publishEvent(any(DaprProtos.PublishEventRequest.class))) - .thenAnswer(c -> { - settableFuture.set(Empty.newBuilder().build()); - return settableFuture; - }); + .thenAnswer(c -> { + settableFuture.set(Empty.newBuilder().build()); + return settableFuture; + }); adapter.publishEvent("pubsubname", "topic", "object"); // Do not call block() on the mono above, so nothing should happen. assertFalse(callback.wasCalled); @@ -158,7 +158,7 @@ public void publishEventObjectTest() { MockCallback callback = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFuture, callback, directExecutor()); when(client.publishEvent(any(DaprProtos.PublishEventRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); MyObject event = new MyObject(1, "Event"); Mono result = adapter.publishEvent("pubsubname", "topic", event); settableFuture.set(Empty.newBuilder().build()); @@ -192,7 +192,7 @@ public void invokeBindingSerializeException() throws IOException { adapter = new DaprClientGrpc(closeable, client, mockSerializer, new DefaultObjectSerializer()); SettableFuture settableFuture = SettableFuture.create(); when(client.invokeBinding(any(DaprProtos.InvokeBindingRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); when(mockSerializer.serialize(any())).thenThrow(IOException.class); Mono result = adapter.invokeBinding("BindingName", "MyOperation", "request".getBytes(), Collections.EMPTY_MAP); result.block(); @@ -201,7 +201,7 @@ public void invokeBindingSerializeException() throws IOException { @Test(expected = RuntimeException.class) public void invokeBindingExceptionThrownTest() { when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenThrow(RuntimeException.class); + .thenThrow(RuntimeException.class); Mono result = adapter.invokeBinding("BindingName", "MyOperation", "request"); result.block(); } @@ -211,11 +211,11 @@ public void invokeBindingCallbackExceptionThrownTest() { SettableFuture settableFuture = SettableFuture.create(); RuntimeException ex = new RuntimeException("An Exception"); MockCallback callback = - new MockCallback<>(ex); + new MockCallback<>(ex); addCallback(settableFuture, callback, directExecutor()); settableFuture.setException(ex); when(client.invokeBinding(any(DaprProtos.InvokeBindingRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeBinding("BindingName", "MyOperation", "request"); result.block(); } @@ -224,11 +224,11 @@ public void invokeBindingCallbackExceptionThrownTest() { public void invokeBindingTest() throws IOException { SettableFuture settableFuture = SettableFuture.create(); DaprProtos.InvokeBindingResponse.Builder responseBuilder = - DaprProtos.InvokeBindingResponse.newBuilder().setData(serialize("OK")); + DaprProtos.InvokeBindingResponse.newBuilder().setData(serialize("OK")); MockCallback callback = new MockCallback<>(responseBuilder.build()); addCallback(settableFuture, callback, directExecutor()); when(client.invokeBinding(any(DaprProtos.InvokeBindingRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeBinding("BindingName", "MyOperation", "request"); settableFuture.set(responseBuilder.build()); result.block(); @@ -239,11 +239,11 @@ public void invokeBindingTest() throws IOException { public void invokeBindingByteArrayTest() throws IOException { SettableFuture settableFuture = SettableFuture.create(); DaprProtos.InvokeBindingResponse.Builder responseBuilder = - DaprProtos.InvokeBindingResponse.newBuilder().setData(ByteString.copyFrom("OK".getBytes())); + DaprProtos.InvokeBindingResponse.newBuilder().setData(ByteString.copyFrom("OK".getBytes())); MockCallback callback = new MockCallback<>(responseBuilder.build()); addCallback(settableFuture, callback, directExecutor()); when(client.invokeBinding(any(DaprProtos.InvokeBindingRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeBinding("BindingName", "MyOperation", "request".getBytes(), Collections.EMPTY_MAP); settableFuture.set(responseBuilder.build()); assertEquals("OK", new String(result.block(), StandardCharsets.UTF_8)); @@ -254,11 +254,11 @@ public void invokeBindingByteArrayTest() throws IOException { public void invokeBindingObjectTest() throws IOException { SettableFuture settableFuture = SettableFuture.create(); DaprProtos.InvokeBindingResponse.Builder responseBuilder = - DaprProtos.InvokeBindingResponse.newBuilder().setData(serialize("OK")); + DaprProtos.InvokeBindingResponse.newBuilder().setData(serialize("OK")); MockCallback callback = new MockCallback<>(responseBuilder.build()); addCallback(settableFuture, callback, directExecutor()); when(client.invokeBinding(any(DaprProtos.InvokeBindingRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); MyObject event = new MyObject(1, "Event"); Mono result = adapter.invokeBinding("BindingName", "MyOperation", event); settableFuture.set(responseBuilder.build()); @@ -270,11 +270,11 @@ public void invokeBindingObjectTest() throws IOException { public void invokeBindingResponseObjectTest() throws IOException { SettableFuture settableFuture = SettableFuture.create(); DaprProtos.InvokeBindingResponse.Builder responseBuilder = - DaprProtos.InvokeBindingResponse.newBuilder().setData(serialize("OK")); + DaprProtos.InvokeBindingResponse.newBuilder().setData(serialize("OK")); MockCallback callback = new MockCallback<>(responseBuilder.build()); addCallback(settableFuture, callback, directExecutor()); when(client.invokeBinding(any(DaprProtos.InvokeBindingRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); MyObject event = new MyObject(1, "Event"); Mono result = adapter.invokeBinding("BindingName", "MyOperation", event, String.class); settableFuture.set(responseBuilder.build()); @@ -286,11 +286,11 @@ public void invokeBindingResponseObjectTest() throws IOException { public void invokeBindingResponseObjectTypeRefTest() throws IOException { SettableFuture settableFuture = SettableFuture.create(); DaprProtos.InvokeBindingResponse.Builder responseBuilder = - DaprProtos.InvokeBindingResponse.newBuilder().setData(serialize("OK")); + DaprProtos.InvokeBindingResponse.newBuilder().setData(serialize("OK")); MockCallback callback = new MockCallback<>(responseBuilder.build()); addCallback(settableFuture, callback, directExecutor()); when(client.invokeBinding(any(DaprProtos.InvokeBindingRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); MyObject event = new MyObject(1, "Event"); Mono result = adapter.invokeBinding("BindingName", "MyOperation", event, TypeRef.get(String.class)); settableFuture.set(responseBuilder.build()); @@ -304,10 +304,10 @@ public void invokeBindingObjectNoHotMono() { MockCallback callback = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFuture, callback, directExecutor()); when(client.invokeBinding(any(DaprProtos.InvokeBindingRequest.class))) - .thenAnswer(c -> { - settableFuture.set(Empty.newBuilder().build()); - return settableFuture; - }); + .thenAnswer(c -> { + settableFuture.set(Empty.newBuilder().build()); + return settableFuture; + }); MyObject event = new MyObject(1, "Event"); adapter.invokeBinding("BindingName", "MyOperation", event); // Do not call block() on mono above, so nothing should happen. @@ -317,7 +317,7 @@ public void invokeBindingObjectNoHotMono() { @Test(expected = RuntimeException.class) public void invokeServiceVoidExceptionThrownTest() { when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenThrow(RuntimeException.class); + .thenThrow(RuntimeException.class); Mono result = adapter.invokeService("appId", "method", "request", HttpExtension.NONE); result.block(); } @@ -326,7 +326,7 @@ public void invokeServiceVoidExceptionThrownTest() { public void invokeServiceIllegalArgumentExceptionThrownTest() { SettableFuture settableFuture = SettableFuture.create(); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); // HttpExtension cannot be null Mono result = adapter.invokeService("appId", "method", "request", null); result.block(); @@ -335,7 +335,7 @@ public void invokeServiceIllegalArgumentExceptionThrownTest() { @Test(expected = RuntimeException.class) public void invokeServiceEmptyRequestVoidExceptionThrownTest() { when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenThrow(RuntimeException.class); + .thenThrow(RuntimeException.class); Mono result = adapter.invokeService("appId", "method", HttpExtension.NONE, (Map)null); result.block(); } @@ -348,7 +348,7 @@ public void invokeServiceVoidCallbackExceptionThrownTest() { addCallback(settableFuture, callback, directExecutor()); settableFuture.setException(ex); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeService("appId", "method", "request", HttpExtension.NONE); result.block(); } @@ -361,7 +361,7 @@ public void invokeServiceVoidTest() throws Exception { .setData(getAny("Value")).build()); addCallback(settableFuture, callback, directExecutor()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeService("appId", "method", "request", HttpExtension.NONE); settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny("Value")).build()); result.block(); @@ -376,7 +376,7 @@ public void invokeServiceVoidObjectTest() throws Exception { .setData(getAny("Value")).build()); addCallback(settableFuture, callback, directExecutor()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); MyObject request = new MyObject(1, "Event"); Mono result = adapter.invokeService("appId", "method", request, HttpExtension.NONE); settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny("Value")).build()); @@ -387,7 +387,7 @@ public void invokeServiceVoidObjectTest() throws Exception { @Test(expected = RuntimeException.class) public void invokeServiceExceptionThrownTest() { when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenThrow(RuntimeException.class); + .thenThrow(RuntimeException.class); Mono result = adapter.invokeService("appId", "method", "request", HttpExtension.NONE, null, String.class); result.block(); } @@ -395,7 +395,7 @@ public void invokeServiceExceptionThrownTest() { @Test(expected = RuntimeException.class) public void invokeServiceNoRequestClassExceptionThrownTest() { when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenThrow(RuntimeException.class); + .thenThrow(RuntimeException.class); Mono result = adapter.invokeService("appId", "method", HttpExtension.NONE, (Map)null, String.class); result.block(); } @@ -403,7 +403,7 @@ public void invokeServiceNoRequestClassExceptionThrownTest() { @Test(expected = RuntimeException.class) public void invokeServiceNoRequestTypeRefExceptionThrownTest() { when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenThrow(RuntimeException.class); + .thenThrow(RuntimeException.class); Mono result = adapter.invokeService("appId", "method", HttpExtension.NONE, (Map)null, TypeRef.STRING); result.block(); } @@ -415,7 +415,7 @@ public void invokeServiceCallbackExceptionThrownTest() { MockCallback callback = new MockCallback<>(ex); addCallback(settableFuture, callback, directExecutor()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeService("appId", "method", "request", HttpExtension.NONE, null, String.class); settableFuture.setException(ex); result.block(); @@ -427,17 +427,17 @@ public void invokeServiceWithHttpExtensionTest() throws IOException { put("test", "1"); }}); CommonProtos.InvokeRequest message = CommonProtos.InvokeRequest.newBuilder() - .setMethod("method") - .setData(getAny("request")) - .setContentType("application/json") - .setHttpExtension(CommonProtos.HTTPExtension.newBuilder() - .setVerb(CommonProtos.HTTPExtension.Verb.GET) - .putQuerystring("test", "1").build()) - .build(); + .setMethod("method") + .setData(getAny("request")) + .setContentType("application/json") + .setHttpExtension(CommonProtos.HTTPExtension.newBuilder() + .setVerb(CommonProtos.HTTPExtension.Verb.GET) + .putQuerystring("test", "1").build()) + .build(); DaprProtos.InvokeServiceRequest request = DaprProtos.InvokeServiceRequest.newBuilder() - .setId("appId") - .setMessage(message) - .build(); + .setId("appId") + .setMessage(message) + .build(); String expected = "Value"; SettableFuture settableFuture = SettableFuture.create(); MockCallback callback = new MockCallback<>(CommonProtos.InvokeResponse.newBuilder() @@ -445,7 +445,7 @@ public void invokeServiceWithHttpExtensionTest() throws IOException { addCallback(settableFuture, callback, directExecutor()); settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny(expected)).build()); when(client.invokeService(eq(request))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeService("appId", "method", "request", httpExtension, null, String.class); String strOutput = result.block(); assertEquals(expected, strOutput); @@ -460,7 +460,7 @@ public void invokeServiceTest() throws Exception { addCallback(settableFuture, callback, directExecutor()); settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny(expected)).build()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeService("appId", "method", "request", HttpExtension.NONE, null, String.class); String strOutput = result.block(); assertEquals(expected, strOutput); @@ -475,7 +475,7 @@ public void invokeServiceObjectTest() throws Exception { addCallback(settableFuture, callback, directExecutor()); settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny(object)).build()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeService("appId", "method", "request", HttpExtension.NONE, null, MyObject.class); MyObject resultObject = result.block(); assertEquals(object.id, resultObject.id); @@ -485,7 +485,7 @@ public void invokeServiceObjectTest() throws Exception { @Test(expected = RuntimeException.class) public void invokeServiceNoRequestBodyExceptionThrownTest() { when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenThrow(RuntimeException.class); + .thenThrow(RuntimeException.class); Mono result = adapter.invokeService("appId", "method", (Object)null, HttpExtension.NONE, String.class); result.block(); } @@ -497,7 +497,7 @@ public void invokeServiceNoRequestCallbackExceptionThrownTest() { MockCallback callback = new MockCallback<>(ex); addCallback(settableFuture, callback, directExecutor()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeService("appId", "method", (Object)null, HttpExtension.NONE, String.class); settableFuture.setException(ex); result.block(); @@ -513,7 +513,7 @@ public void invokeServiceNoRequestBodyTest() throws Exception { addCallback(settableFuture, callback, directExecutor()); settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny(expected)).build()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeService("appId", "method", (Object)null, HttpExtension.NONE, String.class); String strOutput = result.block(); assertEquals(expected, strOutput); @@ -529,7 +529,7 @@ public void invokeServiceNoRequestBodyObjectTest() throws Exception { addCallback(settableFuture, callback, directExecutor()); settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny(object)).build()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeService("appId", "method", (Object)null, HttpExtension.NONE, MyObject.class); MyObject resultObject = result.block(); assertEquals(object.id, resultObject.id); @@ -539,7 +539,7 @@ public void invokeServiceNoRequestBodyObjectTest() throws Exception { @Test(expected = RuntimeException.class) public void invokeServiceByteRequestExceptionThrownTest() throws IOException { when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenThrow(RuntimeException.class); + .thenThrow(RuntimeException.class); String request = "Request"; byte[] byteRequest = serializer.serialize(request); Mono result = adapter.invokeService("appId", "method", byteRequest, HttpExtension.NONE, byte[].class); @@ -553,11 +553,11 @@ public void invokeServiceByteRequestCallbackExceptionThrownTest() throws IOExcep MockCallback callback = new MockCallback<>(ex); addCallback(settableFuture, callback, directExecutor()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); String request = "Request"; byte[] byteRequest = serializer.serialize(request); Mono result = - adapter.invokeService("appId", "method", byteRequest, HttpExtension.NONE,(HashMap) null); + adapter.invokeService("appId", "method", byteRequest, HttpExtension.NONE,(HashMap) null); settableFuture.setException(ex); result.block(); } @@ -571,11 +571,11 @@ public void invokeByteRequestServiceTest() throws Exception { addCallback(settableFuture, callback, directExecutor()); settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny(expected)).build()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); String request = "Request"; byte[] byteRequest = serializer.serialize(request); Mono result = adapter.invokeService( - "appId", "method", byteRequest, HttpExtension.NONE, (HashMap) null); + "appId", "method", byteRequest, HttpExtension.NONE, (HashMap) null); byte[] byteOutput = result.block(); String strOutput = serializer.deserialize(byteOutput, String.class); assertEquals(expected, strOutput); @@ -590,7 +590,7 @@ public void invokeServiceByteRequestObjectTest() throws Exception { addCallback(settableFuture, callback, directExecutor()); settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny(resultObj)).build()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); String request = "Request"; byte[] byteRequest = serializer.serialize(request); Mono result = adapter.invokeService("appId", "method", byteRequest, HttpExtension.NONE, byte[].class); @@ -601,7 +601,7 @@ public void invokeServiceByteRequestObjectTest() throws Exception { @Test(expected = RuntimeException.class) public void invokeServiceNoRequestNoClassBodyExceptionThrownTest() { when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenThrow(RuntimeException.class); + .thenThrow(RuntimeException.class); Mono result = adapter.invokeService("appId", "method", (Object)null, HttpExtension.NONE); result.block(); } @@ -613,7 +613,7 @@ public void invokeServiceNoRequestNoClassCallbackExceptionThrownTest() { MockCallback callback = new MockCallback<>(ex); addCallback(settableFuture, callback, directExecutor()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeService("appId", "method", (Object)null, HttpExtension.NONE); settableFuture.setException(ex); result.block(); @@ -627,7 +627,7 @@ public void invokeServiceNoRequestNoClassBodyTest() throws Exception { .setData(getAny(expected)).build()); addCallback(settableFuture, callback, directExecutor()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeService("appId", "method", (Object)null, HttpExtension.NONE); settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny(expected)).build()); result.block(); @@ -642,10 +642,10 @@ public void invokeServiceNoRequestNoHotMono() throws Exception { .setData(getAny(expected)).build()); addCallback(settableFuture, callback, directExecutor()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenAnswer(c -> { - settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny(expected)).build()); - return settableFuture; - }); + .thenAnswer(c -> { + settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny(expected)).build()); + return settableFuture; + }); adapter.invokeService("appId", "method", (Object)null, HttpExtension.NONE); // Do not call block() on mono above, so nothing should happen. assertFalse(callback.wasCalled); @@ -661,7 +661,7 @@ public void invokeServiceNoRequestNoClassBodyObjectTest() throws Exception { addCallback(settableFuture, callback, directExecutor()); settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny(resultObj)).build()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeService("appId", "method", (Object)null, HttpExtension.NONE); result.block(); assertTrue(callback.wasCalled); @@ -701,10 +701,10 @@ public void getStateCallbackExceptionThrownTest() { SettableFuture settableFuture = SettableFuture.create(); RuntimeException ex = new RuntimeException("An Exception"); MockCallback callback = - new MockCallback<>(ex); + new MockCallback<>(ex); addCallback(settableFuture, callback, directExecutor()); when(client.getState(any(io.dapr.v1.DaprProtos.GetStateRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); State key = buildStateKey(null, "Key1", "ETag1", null); Mono> result = adapter.getState(STATE_STORE_NAME, key, String.class); settableFuture.setException(ex); @@ -716,13 +716,13 @@ public void getStateStringValueNoOptionsTest() throws IOException { String etag = "ETag1"; String key = "key1"; String expectedValue = "Expected state"; - State expectedState = buildStateKey(expectedValue, key, etag, null); + State expectedState = buildStateKey(expectedValue, key, etag, new HashMap<>(), null); DaprProtos.GetStateResponse responseEnvelope = buildGetStateResponse(expectedValue, etag); SettableFuture settableFuture = SettableFuture.create(); MockCallback callback = new MockCallback<>(responseEnvelope); addCallback(settableFuture, callback, directExecutor()); when(client.getState(any(io.dapr.v1.DaprProtos.GetStateRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); State keyRequest = buildStateKey(null, key, etag, null); Mono> result = adapter.getState(STATE_STORE_NAME, keyRequest, String.class); settableFuture.set(responseEnvelope); @@ -742,10 +742,10 @@ public void getStateStringValueNoHotMono() throws IOException { MockCallback callback = new MockCallback<>(responseEnvelope); addCallback(settableFuture, callback, directExecutor()); when(client.getState(any(io.dapr.v1.DaprProtos.GetStateRequest.class))) - .thenAnswer(c -> { - settableFuture.set(responseEnvelope); - return settableFuture; - }); + .thenAnswer(c -> { + settableFuture.set(responseEnvelope); + return settableFuture; + }); State keyRequest = buildStateKey(null, key, etag, null); adapter.getState(STATE_STORE_NAME, keyRequest, String.class); // block() on the mono above is not called, so nothing should happen. @@ -758,17 +758,17 @@ public void getStateObjectValueWithOptionsTest() throws IOException { String key = "key1"; MyObject expectedValue = new MyObject(1, "The Value"); StateOptions options = buildStateOptions(StateOptions.Consistency.STRONG, StateOptions.Concurrency.FIRST_WRITE); - State expectedState = buildStateKey(expectedValue, key, etag, options); + State expectedState = buildStateKey(expectedValue, key, etag, new HashMap<>(), options); DaprProtos.GetStateResponse responseEnvelope = DaprProtos.GetStateResponse.newBuilder() - .setData(serialize(expectedValue)) - .setEtag(etag) - .build(); + .setData(serialize(expectedValue)) + .setEtag(etag) + .build(); State keyRequest = buildStateKey(null, key, etag, options); SettableFuture settableFuture = SettableFuture.create(); MockCallback callback = new MockCallback<>(responseEnvelope); addCallback(settableFuture, callback, directExecutor()); when(client.getState(any(io.dapr.v1.DaprProtos.GetStateRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono> result = adapter.getState(STATE_STORE_NAME, keyRequest, MyObject.class); settableFuture.set(responseEnvelope); State res = result.block(); @@ -784,11 +784,11 @@ public void getStateObjectValueWithMetadataTest() throws IOException { StateOptions options = buildStateOptions(StateOptions.Consistency.STRONG, StateOptions.Concurrency.FIRST_WRITE); Map metadata = new HashMap<>(); metadata.put("key_1", "val_1"); - State expectedState = buildStateKey(expectedValue, key, etag, options); + State expectedState = buildStateKey(expectedValue, key, etag, new HashMap<>(), options); DaprProtos.GetStateResponse responseEnvelope = DaprProtos.GetStateResponse.newBuilder() - .setData(serialize(expectedValue)) - .setEtag(etag) - .build(); + .setData(serialize(expectedValue)) + .setEtag(etag) + .build(); GetStateRequestBuilder builder = new GetStateRequestBuilder(STATE_STORE_NAME, key); builder.withMetadata(metadata).withEtag(etag).withStateOptions(options); GetStateRequest request = builder.build(); @@ -796,7 +796,7 @@ public void getStateObjectValueWithMetadataTest() throws IOException { MockCallback callback = new MockCallback<>(responseEnvelope); addCallback(settableFuture, callback, directExecutor()); when(client.getState(any(io.dapr.v1.DaprProtos.GetStateRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono>> result = adapter.getState(request, TypeRef.get(MyObject.class)); settableFuture.set(responseEnvelope); Response> res = result.block(); @@ -810,17 +810,17 @@ public void getStateObjectValueWithOptionsNoConcurrencyTest() throws IOException String key = "key1"; MyObject expectedValue = new MyObject(1, "The Value"); StateOptions options = new StateOptions(null, StateOptions.Concurrency.FIRST_WRITE); - State expectedState = buildStateKey(expectedValue, key, etag, options); + State expectedState = buildStateKey(expectedValue, key, etag, new HashMap<>(), options); DaprProtos.GetStateResponse responseEnvelope = DaprProtos.GetStateResponse.newBuilder() - .setData(serialize(expectedValue)) - .setEtag(etag) - .build(); + .setData(serialize(expectedValue)) + .setEtag(etag) + .build(); State keyRequest = buildStateKey(null, key, etag, options); SettableFuture settableFuture = SettableFuture.create(); MockCallback callback = new MockCallback<>(responseEnvelope); addCallback(settableFuture, callback, directExecutor()); when(client.getState(any(io.dapr.v1.DaprProtos.GetStateRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono> result = adapter.getState(STATE_STORE_NAME, keyRequest, MyObject.class); settableFuture.set(responseEnvelope); assertEquals(expectedState, result.block()); @@ -848,32 +848,32 @@ public void getStatesIllegalArgumentExceptionTest() { }); // negative parallelism GetStatesRequest req = new GetStatesRequestBuilder(STATE_STORE_NAME, Collections.singletonList("100")) - .withParallelism(-1) - .build(); + .withParallelism(-1) + .build(); assertThrows(IllegalArgumentException.class, () -> adapter.getStates(req, TypeRef.BOOLEAN).block()); } @Test public void getStatesString() throws IOException { DaprProtos.GetBulkStateResponse responseEnvelope = DaprProtos.GetBulkStateResponse.newBuilder() - .addItems(DaprProtos.BulkStateItem.newBuilder() - .setData(serialize("hello world")) - .setKey("100") - .setEtag("1") - .build()) - .addItems(DaprProtos.BulkStateItem.newBuilder() - .setKey("200") - .setError("not found") - .build()) - .build(); + .addItems(DaprProtos.BulkStateItem.newBuilder() + .setData(serialize("hello world")) + .setKey("100") + .setEtag("1") + .build()) + .addItems(DaprProtos.BulkStateItem.newBuilder() + .setKey("200") + .setError("not found") + .build()) + .build(); SettableFuture settableFuture = SettableFuture.create(); MockCallback callback = new MockCallback<>(responseEnvelope); addCallback(settableFuture, callback, directExecutor()); when(client.getBulkState(any(DaprProtos.GetBulkStateRequest.class))) - .thenAnswer(c -> { - settableFuture.set(responseEnvelope); - return settableFuture; - }); + .thenAnswer(c -> { + settableFuture.set(responseEnvelope); + return settableFuture; + }); List> result = adapter.getStates(STATE_STORE_NAME, Arrays.asList("100", "200"), String.class).block(); assertTrue(callback.wasCalled); @@ -891,24 +891,24 @@ public void getStatesString() throws IOException { @Test public void getStatesInteger() throws IOException { DaprProtos.GetBulkStateResponse responseEnvelope = DaprProtos.GetBulkStateResponse.newBuilder() - .addItems(DaprProtos.BulkStateItem.newBuilder() - .setData(serialize(1234)) - .setKey("100") - .setEtag("1") - .build()) - .addItems(DaprProtos.BulkStateItem.newBuilder() - .setKey("200") - .setError("not found") - .build()) - .build(); + .addItems(DaprProtos.BulkStateItem.newBuilder() + .setData(serialize(1234)) + .setKey("100") + .setEtag("1") + .build()) + .addItems(DaprProtos.BulkStateItem.newBuilder() + .setKey("200") + .setError("not found") + .build()) + .build(); SettableFuture settableFuture = SettableFuture.create(); MockCallback callback = new MockCallback<>(responseEnvelope); addCallback(settableFuture, callback, directExecutor()); when(client.getBulkState(any(DaprProtos.GetBulkStateRequest.class))) - .thenAnswer(c -> { - settableFuture.set(responseEnvelope); - return settableFuture; - }); + .thenAnswer(c -> { + settableFuture.set(responseEnvelope); + return settableFuture; + }); List> result = adapter.getStates(STATE_STORE_NAME, Arrays.asList("100", "200"), int.class).block(); assertTrue(callback.wasCalled); @@ -926,24 +926,24 @@ public void getStatesInteger() throws IOException { @Test public void getStatesBoolean() throws IOException { DaprProtos.GetBulkStateResponse responseEnvelope = DaprProtos.GetBulkStateResponse.newBuilder() - .addItems(DaprProtos.BulkStateItem.newBuilder() - .setData(serialize(true)) - .setKey("100") - .setEtag("1") - .build()) - .addItems(DaprProtos.BulkStateItem.newBuilder() - .setKey("200") - .setError("not found") - .build()) - .build(); + .addItems(DaprProtos.BulkStateItem.newBuilder() + .setData(serialize(true)) + .setKey("100") + .setEtag("1") + .build()) + .addItems(DaprProtos.BulkStateItem.newBuilder() + .setKey("200") + .setError("not found") + .build()) + .build(); SettableFuture settableFuture = SettableFuture.create(); MockCallback callback = new MockCallback<>(responseEnvelope); addCallback(settableFuture, callback, directExecutor()); when(client.getBulkState(any(DaprProtos.GetBulkStateRequest.class))) - .thenAnswer(c -> { - settableFuture.set(responseEnvelope); - return settableFuture; - }); + .thenAnswer(c -> { + settableFuture.set(responseEnvelope); + return settableFuture; + }); List> result = adapter.getStates(STATE_STORE_NAME, Arrays.asList("100", "200"), boolean.class).block(); assertTrue(callback.wasCalled); @@ -961,24 +961,24 @@ public void getStatesBoolean() throws IOException { @Test public void getStatesByteArray() throws IOException { DaprProtos.GetBulkStateResponse responseEnvelope = DaprProtos.GetBulkStateResponse.newBuilder() - .addItems(DaprProtos.BulkStateItem.newBuilder() - .setData(serialize(new byte[]{1, 2, 3})) - .setKey("100") - .setEtag("1") - .build()) - .addItems(DaprProtos.BulkStateItem.newBuilder() - .setKey("200") - .setError("not found") - .build()) - .build(); + .addItems(DaprProtos.BulkStateItem.newBuilder() + .setData(serialize(new byte[]{1, 2, 3})) + .setKey("100") + .setEtag("1") + .build()) + .addItems(DaprProtos.BulkStateItem.newBuilder() + .setKey("200") + .setError("not found") + .build()) + .build(); SettableFuture settableFuture = SettableFuture.create(); MockCallback callback = new MockCallback<>(responseEnvelope); addCallback(settableFuture, callback, directExecutor()); when(client.getBulkState(any(DaprProtos.GetBulkStateRequest.class))) - .thenAnswer(c -> { - settableFuture.set(responseEnvelope); - return settableFuture; - }); + .thenAnswer(c -> { + settableFuture.set(responseEnvelope); + return settableFuture; + }); List> result = adapter.getStates(STATE_STORE_NAME, Arrays.asList("100", "200"), byte[].class).block(); assertTrue(callback.wasCalled); @@ -997,24 +997,24 @@ public void getStatesByteArray() throws IOException { public void getStatesObject() throws IOException { MyObject object = new MyObject(1, "Event"); DaprProtos.GetBulkStateResponse responseEnvelope = DaprProtos.GetBulkStateResponse.newBuilder() - .addItems(DaprProtos.BulkStateItem.newBuilder() - .setData(serialize(object)) - .setKey("100") - .setEtag("1") - .build()) - .addItems(DaprProtos.BulkStateItem.newBuilder() - .setKey("200") - .setError("not found") - .build()) - .build(); + .addItems(DaprProtos.BulkStateItem.newBuilder() + .setData(serialize(object)) + .setKey("100") + .setEtag("1") + .build()) + .addItems(DaprProtos.BulkStateItem.newBuilder() + .setKey("200") + .setError("not found") + .build()) + .build(); SettableFuture settableFuture = SettableFuture.create(); MockCallback callback = new MockCallback<>(responseEnvelope); addCallback(settableFuture, callback, directExecutor()); when(client.getBulkState(any(DaprProtos.GetBulkStateRequest.class))) - .thenAnswer(c -> { - settableFuture.set(responseEnvelope); - return settableFuture; - }); + .thenAnswer(c -> { + settableFuture.set(responseEnvelope); + return settableFuture; + }); List> result = adapter.getStates(STATE_STORE_NAME, Arrays.asList("100", "200"), MyObject.class).block(); assertTrue(callback.wasCalled); @@ -1065,7 +1065,7 @@ public void deleteStateCallbackExcpetionThrownTest() { MockCallback callback = new MockCallback<>(ex); addCallback(settableFuture, callback, directExecutor()); when(client.deleteState(any(io.dapr.v1.DaprProtos.DeleteStateRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); State key = buildStateKey(null, "Key1", "ETag1", null); Mono result = adapter.deleteState(STATE_STORE_NAME, key.getKey(), key.getEtag(), key.getOptions()); settableFuture.setException(ex); @@ -1080,10 +1080,10 @@ public void deleteStateNoOptionsTest() { MockCallback callback = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFuture, callback, directExecutor()); when(client.deleteState(any(io.dapr.v1.DaprProtos.DeleteStateRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); State stateKey = buildStateKey(null, key, etag, null); Mono result = adapter.deleteState(STATE_STORE_NAME, stateKey.getKey(), stateKey.getEtag(), - stateKey.getOptions()); + stateKey.getOptions()); settableFuture.set(Empty.newBuilder().build()); result.block(); assertTrue(callback.wasCalled); @@ -1098,10 +1098,10 @@ public void deleteStateTest() { MockCallback callback = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFuture, callback, directExecutor()); when(client.deleteState(any(io.dapr.v1.DaprProtos.DeleteStateRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); State stateKey = buildStateKey(null, key, etag, options); Mono result = adapter.deleteState(STATE_STORE_NAME, stateKey.getKey(), stateKey.getEtag(), - stateKey.getOptions()); + stateKey.getOptions()); settableFuture.set(Empty.newBuilder().build()); result.block(); assertTrue(callback.wasCalled); @@ -1118,7 +1118,7 @@ public void deleteStateWithMetadata() { MockCallback callback = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFuture, callback, directExecutor()); when(client.deleteState(any(io.dapr.v1.DaprProtos.DeleteStateRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); DeleteStateRequestBuilder builder = new DeleteStateRequestBuilder(STATE_STORE_NAME, key); builder.withEtag(etag).withStateOptions(options).withMetadata(metadata); DeleteStateRequest request = builder.build(); @@ -1137,13 +1137,13 @@ public void deleteStateTestNoHotMono() { MockCallback callback = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFuture, callback, directExecutor()); when(client.deleteState(any(io.dapr.v1.DaprProtos.DeleteStateRequest.class))) - .thenAnswer(c -> { - settableFuture.set(Empty.newBuilder().build()); - return settableFuture; - }); + .thenAnswer(c -> { + settableFuture.set(Empty.newBuilder().build()); + return settableFuture; + }); State stateKey = buildStateKey(null, key, etag, options); Mono result = adapter.deleteState(STATE_STORE_NAME, stateKey.getKey(), stateKey.getEtag(), - stateKey.getOptions()); + stateKey.getOptions()); // Do not call result.block(), so nothing should happen. assertFalse(callback.wasCalled); } @@ -1157,10 +1157,10 @@ public void deleteStateNoConsistencyTest() { MockCallback callback = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFuture, callback, directExecutor()); when(client.deleteState(any(io.dapr.v1.DaprProtos.DeleteStateRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); State stateKey = buildStateKey(null, key, etag, options); Mono result = adapter.deleteState(STATE_STORE_NAME, stateKey.getKey(), stateKey.getEtag(), - stateKey.getOptions()); + stateKey.getOptions()); settableFuture.set(Empty.newBuilder().build()); result.block(); assertTrue(callback.wasCalled); @@ -1175,10 +1175,10 @@ public void deleteStateNoConcurrencyTest() { MockCallback callback = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFuture, callback, directExecutor()); when(client.deleteState(any(io.dapr.v1.DaprProtos.DeleteStateRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); State stateKey = buildStateKey(null, key, etag, options); Mono result = adapter.deleteState(STATE_STORE_NAME, stateKey.getKey(), stateKey.getEtag(), - stateKey.getOptions()); + stateKey.getOptions()); settableFuture.set(Empty.newBuilder().build()); result.block(); assertTrue(callback.wasCalled); @@ -1188,8 +1188,8 @@ public void deleteStateNoConcurrencyTest() { public void executeTransactionIllegalArgumentExceptionTest() { State key = buildStateKey(null, "Key1", "ETag1", null); TransactionalStateOperation upsertOperation = new TransactionalStateOperation<>( - TransactionalStateOperation.OperationType.UPSERT, - key); + TransactionalStateOperation.OperationType.UPSERT, + key); assertThrows(IllegalArgumentException.class, () -> { // empty state store name adapter.executeTransaction("", Collections.singletonList(upsertOperation)).block(); @@ -1210,15 +1210,15 @@ public void executeTransactionSerializerExceptionTest() throws IOException { StateOptions options = buildStateOptions(StateOptions.Consistency.STRONG, null); SettableFuture settableFuture = SettableFuture.create(); when(client.executeStateTransaction(any(DaprProtos.ExecuteStateTransactionRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); when(mockSerializer.serialize(any())).thenThrow(IOException.class); State stateKey = buildStateKey(data, key, etag, options); TransactionalStateOperation upsertOperation = new TransactionalStateOperation<>( - TransactionalStateOperation.OperationType.UPSERT, - stateKey); + TransactionalStateOperation.OperationType.UPSERT, + stateKey); ExecuteStateTransactionRequest request = new ExecuteStateTransactionRequestBuilder(STATE_STORE_NAME) - .withTransactionalStates(upsertOperation) - .build(); + .withTransactionalStates(upsertOperation) + .build(); Mono> result = adapter.executeTransaction(request); result.block(); } @@ -1233,20 +1233,20 @@ public void executeTransactionWithMetadataTest() { MockCallback callback = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFuture, callback, directExecutor()); when(client.executeStateTransaction(any(DaprProtos.ExecuteStateTransactionRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); State stateKey = buildStateKey(data, key, etag, options); TransactionalStateOperation upsertOperation = new TransactionalStateOperation<>( - TransactionalStateOperation.OperationType.UPSERT, - stateKey); + TransactionalStateOperation.OperationType.UPSERT, + stateKey); TransactionalStateOperation deleteOperation = new TransactionalStateOperation<>( - TransactionalStateOperation.OperationType.DELETE, - new State<>("testKey")); + TransactionalStateOperation.OperationType.DELETE, + new State<>("testKey")); Map metadata = new HashMap<>(); metadata.put("testKey", "testValue"); ExecuteStateTransactionRequest request = new ExecuteStateTransactionRequestBuilder(STATE_STORE_NAME) - .withTransactionalStates(upsertOperation, deleteOperation) - .withMetadata(metadata) - .build(); + .withTransactionalStates(upsertOperation, deleteOperation) + .withMetadata(metadata) + .build(); Mono> result = adapter.executeTransaction(request); settableFuture.set(Empty.newBuilder().build()); result.block(); @@ -1263,14 +1263,14 @@ public void executeTransactionTest() { MockCallback callback = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFuture, callback, directExecutor()); when(client.executeStateTransaction(any(DaprProtos.ExecuteStateTransactionRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); State stateKey = buildStateKey(data, key, etag, options); TransactionalStateOperation upsertOperation = new TransactionalStateOperation<>( - TransactionalStateOperation.OperationType.UPSERT, - stateKey); + TransactionalStateOperation.OperationType.UPSERT, + stateKey); TransactionalStateOperation deleteOperation = new TransactionalStateOperation<>( - TransactionalStateOperation.OperationType.DELETE, - new State<>("testKey") + TransactionalStateOperation.OperationType.DELETE, + new State<>("testKey") ); Mono result = adapter.executeTransaction(STATE_STORE_NAME, Arrays.asList(upsertOperation, deleteOperation)); settableFuture.set(Empty.newBuilder().build()); @@ -1285,11 +1285,11 @@ public void executeTransactionExceptionThrownTest() { String data = "my data"; StateOptions options = buildStateOptions(StateOptions.Consistency.STRONG, null); when(client.executeStateTransaction(any(DaprProtos.ExecuteStateTransactionRequest.class))) - .thenThrow(RuntimeException.class); + .thenThrow(RuntimeException.class); State stateKey = buildStateKey(data, key, etag, options); TransactionalStateOperation operation = new TransactionalStateOperation<>( - TransactionalStateOperation.OperationType.UPSERT, - stateKey); + TransactionalStateOperation.OperationType.UPSERT, + stateKey); Mono result = adapter.executeTransaction(STATE_STORE_NAME, Collections.singletonList(operation)); result.block(); } @@ -1305,11 +1305,11 @@ public void executeTransactionCallbackExceptionTest() { MockCallback callback = new MockCallback<>(ex); addCallback(settableFuture, callback, directExecutor()); when(client.executeStateTransaction(any(DaprProtos.ExecuteStateTransactionRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); State stateKey = buildStateKey(data, key, etag, options); TransactionalStateOperation operation = new TransactionalStateOperation<>( - TransactionalStateOperation.OperationType.UPSERT, - stateKey); + TransactionalStateOperation.OperationType.UPSERT, + stateKey); Mono result = adapter.executeTransaction(STATE_STORE_NAME, Collections.singletonList(operation)); settableFuture.setException(ex); result.block(); @@ -1453,6 +1453,10 @@ private State buildStateKey(T value, String key, String etag, StateOption return new State<>(value, key, etag, options); } + private State buildStateKey(T value, String key, String etag, Map metadata, StateOptions options) { + return new State<>(value, key, etag, metadata, options); + } + /** * The purpose of this test is to show that it doesn't matter when the client is called, the actual coll to DAPR * will be done when the output Mono response call the Mono.block method. @@ -1480,7 +1484,7 @@ public void getStateDeleteStateThenBlockDeleteThenBlockGet() throws Exception { String expectedValue1 = "Expected state 1"; String key2 = "key2"; String expectedValue2 = "Expected state 2"; - State expectedState1 = buildStateKey(expectedValue1, key1, etag, null); + State expectedState1 = buildStateKey(expectedValue1, key1, etag, new HashMap<>(), null); Map> futuresMap = new HashMap<>(); futuresMap.put(key1, buildFutureGetStateEnvelop(expectedValue1, etag)); futuresMap.put(key2, buildFutureGetStateEnvelop(expectedValue2, etag)); @@ -1495,9 +1499,9 @@ public void getStateDeleteStateThenBlockDeleteThenBlockGet() throws Exception { MockCallback callbackDelete = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFutureDelete, callbackDelete, directExecutor()); when(client.deleteState(any(io.dapr.v1.DaprProtos.DeleteStateRequest.class))) - .thenReturn(settableFutureDelete); + .thenReturn(settableFutureDelete); Mono resultDelete = adapter.deleteState(STATE_STORE_NAME, keyRequest2.getKey(), keyRequest2.getEtag(), - keyRequest2.getOptions()); + keyRequest2.getOptions()); settableFutureDelete.set(Empty.newBuilder().build()); resultDelete.block(); assertTrue(callbackDelete.wasCalled); @@ -1519,13 +1523,13 @@ public void getSecrets() { settableFuture.set(responseEnvelope); when(client.getSecret(any(io.dapr.v1.DaprProtos.GetSecretRequest.class))) - .thenAnswer(context -> { - io.dapr.v1.DaprProtos.GetSecretRequest req = context.getArgument(0); - assertEquals("key", req.getKey()); - assertEquals(SECRET_STORE_NAME, req.getStoreName()); - assertEquals(0, req.getMetadataCount()); - return settableFuture; - }); + .thenAnswer(context -> { + io.dapr.v1.DaprProtos.GetSecretRequest req = context.getArgument(0); + assertEquals("key", req.getKey()); + assertEquals(SECRET_STORE_NAME, req.getStoreName()); + assertEquals(0, req.getMetadataCount()); + return settableFuture; + }); Map result = adapter.getSecret(SECRET_STORE_NAME, "key").block(); @@ -1542,13 +1546,13 @@ public void getSecretsEmptyResponse() { settableFuture.set(responseEnvelope); when(client.getSecret(any(io.dapr.v1.DaprProtos.GetSecretRequest.class))) - .thenAnswer(context -> { - io.dapr.v1.DaprProtos.GetSecretRequest req = context.getArgument(0); - assertEquals("key", req.getKey()); - assertEquals(SECRET_STORE_NAME, req.getStoreName()); - assertEquals(0, req.getMetadataCount()); - return settableFuture; - }); + .thenAnswer(context -> { + io.dapr.v1.DaprProtos.GetSecretRequest req = context.getArgument(0); + assertEquals("key", req.getKey()); + assertEquals(SECRET_STORE_NAME, req.getStoreName()); + assertEquals(0, req.getMetadataCount()); + return settableFuture; + }); Map result = adapter.getSecret(SECRET_STORE_NAME, "key").block(); @@ -1563,13 +1567,13 @@ public void getSecretsException() { settableFuture.setException(new RuntimeException()); when(client.getSecret(any(io.dapr.v1.DaprProtos.GetSecretRequest.class))) - .thenAnswer(context -> { - io.dapr.v1.DaprProtos.GetSecretRequest req = context.getArgument(0); - assertEquals("key", req.getKey()); - assertEquals(SECRET_STORE_NAME, req.getStoreName()); - assertEquals(0, req.getMetadataCount()); - return settableFuture; - }); + .thenAnswer(context -> { + io.dapr.v1.DaprProtos.GetSecretRequest req = context.getArgument(0); + assertEquals("key", req.getKey()); + assertEquals(SECRET_STORE_NAME, req.getStoreName()); + assertEquals(0, req.getMetadataCount()); + return settableFuture; + }); assertThrows(RuntimeException.class, () -> adapter.getSecret(SECRET_STORE_NAME, "key").block()); } @@ -1605,18 +1609,18 @@ public void getSecretsWithMetadata() { settableFuture.set(responseEnvelope); when(client.getSecret(any(io.dapr.v1.DaprProtos.GetSecretRequest.class))) - .thenAnswer(context -> { - io.dapr.v1.DaprProtos.GetSecretRequest req = context.getArgument(0); - assertEquals("key", req.getKey()); - assertEquals(SECRET_STORE_NAME, req.getStoreName()); - assertEquals("metavalue", req.getMetadataMap().get("metakey")); - return settableFuture; - }); + .thenAnswer(context -> { + io.dapr.v1.DaprProtos.GetSecretRequest req = context.getArgument(0); + assertEquals("key", req.getKey()); + assertEquals(SECRET_STORE_NAME, req.getStoreName()); + assertEquals("metavalue", req.getMetadataMap().get("metakey")); + return settableFuture; + }); Map result = adapter.getSecret( - SECRET_STORE_NAME, - "key", - Collections.singletonMap("metakey", "metavalue")).block(); + SECRET_STORE_NAME, + "key", + Collections.singletonMap("metakey", "metavalue")).block(); assertEquals(1, result.size()); assertEquals(expectedValue, result.get(expectedKey)); @@ -1678,15 +1682,15 @@ private SettableFuture buildFutureGetStateEnvel private DaprProtos.GetStateResponse buildGetStateResponse(T value, String etag) throws IOException { return DaprProtos.GetStateResponse.newBuilder() - .setData(serialize(value)) - .setEtag(etag) - .build(); + .setData(serialize(value)) + .setEtag(etag) + .build(); } private DaprProtos.GetSecretResponse buildGetSecretResponse(String key, String value) { return DaprProtos.GetSecretResponse.newBuilder() - .putAllData(Collections.singletonMap(key, value)) - .build(); + .putAllData(Collections.singletonMap(key, value)) + .build(); } private DaprProtos.GetSecretResponse buildGetSecretResponse() { From 9114bbb33c4779ce8ff807a52e03915c06a6b1ba Mon Sep 17 00:00:00 2001 From: Sky Ao Date: Thu, 3 Dec 2020 17:02:00 +0800 Subject: [PATCH 2/6] rollback DaprClientGrpcTest which should be comitted in another PR --- .../io/dapr/client/DaprClientGrpcTest.java | 490 +++++++++--------- 1 file changed, 243 insertions(+), 247 deletions(-) diff --git a/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java b/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java index b2e9c68064..cae8998a0a 100644 --- a/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java +++ b/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java @@ -94,7 +94,7 @@ public void tearDown() throws IOException { @Test(expected = RuntimeException.class) public void publishEventExceptionThrownTest() { when(client.publishEvent(any(DaprProtos.PublishEventRequest.class))) - .thenThrow(RuntimeException.class); + .thenThrow(RuntimeException.class); Mono result = adapter.publishEvent("pubsubname","topic", "object"); result.block(); } @@ -106,7 +106,7 @@ public void publishEventCallbackExceptionThrownTest() { MockCallback callback = new MockCallback<>(ex); addCallback(settableFuture, callback, directExecutor()); when(client.publishEvent(any(DaprProtos.PublishEventRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.publishEvent("pubsubname","topic", "object"); settableFuture.setException(ex); result.block(); @@ -118,7 +118,7 @@ public void publishEventSerializeException() throws IOException { adapter = new DaprClientGrpc(closeable, client, mockSerializer, new DefaultObjectSerializer()); SettableFuture settableFuture = SettableFuture.create(); when(client.publishEvent(any(DaprProtos.PublishEventRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); when(mockSerializer.serialize(any())).thenThrow(IOException.class); Mono result = adapter.publishEvent("pubsubname","topic", "{invalid-json"); result.block(); @@ -130,7 +130,7 @@ public void publishEventTest() { MockCallback callback = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFuture, callback, directExecutor()); when(client.publishEvent(any(DaprProtos.PublishEventRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.publishEvent("pubsubname","topic", "object"); settableFuture.set(Empty.newBuilder().build()); result.block(); @@ -143,10 +143,10 @@ public void publishEventNoHotMono() { MockCallback callback = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFuture, callback, directExecutor()); when(client.publishEvent(any(DaprProtos.PublishEventRequest.class))) - .thenAnswer(c -> { - settableFuture.set(Empty.newBuilder().build()); - return settableFuture; - }); + .thenAnswer(c -> { + settableFuture.set(Empty.newBuilder().build()); + return settableFuture; + }); adapter.publishEvent("pubsubname", "topic", "object"); // Do not call block() on the mono above, so nothing should happen. assertFalse(callback.wasCalled); @@ -158,7 +158,7 @@ public void publishEventObjectTest() { MockCallback callback = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFuture, callback, directExecutor()); when(client.publishEvent(any(DaprProtos.PublishEventRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); MyObject event = new MyObject(1, "Event"); Mono result = adapter.publishEvent("pubsubname", "topic", event); settableFuture.set(Empty.newBuilder().build()); @@ -192,7 +192,7 @@ public void invokeBindingSerializeException() throws IOException { adapter = new DaprClientGrpc(closeable, client, mockSerializer, new DefaultObjectSerializer()); SettableFuture settableFuture = SettableFuture.create(); when(client.invokeBinding(any(DaprProtos.InvokeBindingRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); when(mockSerializer.serialize(any())).thenThrow(IOException.class); Mono result = adapter.invokeBinding("BindingName", "MyOperation", "request".getBytes(), Collections.EMPTY_MAP); result.block(); @@ -201,7 +201,7 @@ public void invokeBindingSerializeException() throws IOException { @Test(expected = RuntimeException.class) public void invokeBindingExceptionThrownTest() { when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenThrow(RuntimeException.class); + .thenThrow(RuntimeException.class); Mono result = adapter.invokeBinding("BindingName", "MyOperation", "request"); result.block(); } @@ -211,11 +211,11 @@ public void invokeBindingCallbackExceptionThrownTest() { SettableFuture settableFuture = SettableFuture.create(); RuntimeException ex = new RuntimeException("An Exception"); MockCallback callback = - new MockCallback<>(ex); + new MockCallback<>(ex); addCallback(settableFuture, callback, directExecutor()); settableFuture.setException(ex); when(client.invokeBinding(any(DaprProtos.InvokeBindingRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeBinding("BindingName", "MyOperation", "request"); result.block(); } @@ -224,11 +224,11 @@ public void invokeBindingCallbackExceptionThrownTest() { public void invokeBindingTest() throws IOException { SettableFuture settableFuture = SettableFuture.create(); DaprProtos.InvokeBindingResponse.Builder responseBuilder = - DaprProtos.InvokeBindingResponse.newBuilder().setData(serialize("OK")); + DaprProtos.InvokeBindingResponse.newBuilder().setData(serialize("OK")); MockCallback callback = new MockCallback<>(responseBuilder.build()); addCallback(settableFuture, callback, directExecutor()); when(client.invokeBinding(any(DaprProtos.InvokeBindingRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeBinding("BindingName", "MyOperation", "request"); settableFuture.set(responseBuilder.build()); result.block(); @@ -239,11 +239,11 @@ public void invokeBindingTest() throws IOException { public void invokeBindingByteArrayTest() throws IOException { SettableFuture settableFuture = SettableFuture.create(); DaprProtos.InvokeBindingResponse.Builder responseBuilder = - DaprProtos.InvokeBindingResponse.newBuilder().setData(ByteString.copyFrom("OK".getBytes())); + DaprProtos.InvokeBindingResponse.newBuilder().setData(ByteString.copyFrom("OK".getBytes())); MockCallback callback = new MockCallback<>(responseBuilder.build()); addCallback(settableFuture, callback, directExecutor()); when(client.invokeBinding(any(DaprProtos.InvokeBindingRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeBinding("BindingName", "MyOperation", "request".getBytes(), Collections.EMPTY_MAP); settableFuture.set(responseBuilder.build()); assertEquals("OK", new String(result.block(), StandardCharsets.UTF_8)); @@ -254,11 +254,11 @@ public void invokeBindingByteArrayTest() throws IOException { public void invokeBindingObjectTest() throws IOException { SettableFuture settableFuture = SettableFuture.create(); DaprProtos.InvokeBindingResponse.Builder responseBuilder = - DaprProtos.InvokeBindingResponse.newBuilder().setData(serialize("OK")); + DaprProtos.InvokeBindingResponse.newBuilder().setData(serialize("OK")); MockCallback callback = new MockCallback<>(responseBuilder.build()); addCallback(settableFuture, callback, directExecutor()); when(client.invokeBinding(any(DaprProtos.InvokeBindingRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); MyObject event = new MyObject(1, "Event"); Mono result = adapter.invokeBinding("BindingName", "MyOperation", event); settableFuture.set(responseBuilder.build()); @@ -270,11 +270,11 @@ public void invokeBindingObjectTest() throws IOException { public void invokeBindingResponseObjectTest() throws IOException { SettableFuture settableFuture = SettableFuture.create(); DaprProtos.InvokeBindingResponse.Builder responseBuilder = - DaprProtos.InvokeBindingResponse.newBuilder().setData(serialize("OK")); + DaprProtos.InvokeBindingResponse.newBuilder().setData(serialize("OK")); MockCallback callback = new MockCallback<>(responseBuilder.build()); addCallback(settableFuture, callback, directExecutor()); when(client.invokeBinding(any(DaprProtos.InvokeBindingRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); MyObject event = new MyObject(1, "Event"); Mono result = adapter.invokeBinding("BindingName", "MyOperation", event, String.class); settableFuture.set(responseBuilder.build()); @@ -286,11 +286,11 @@ public void invokeBindingResponseObjectTest() throws IOException { public void invokeBindingResponseObjectTypeRefTest() throws IOException { SettableFuture settableFuture = SettableFuture.create(); DaprProtos.InvokeBindingResponse.Builder responseBuilder = - DaprProtos.InvokeBindingResponse.newBuilder().setData(serialize("OK")); + DaprProtos.InvokeBindingResponse.newBuilder().setData(serialize("OK")); MockCallback callback = new MockCallback<>(responseBuilder.build()); addCallback(settableFuture, callback, directExecutor()); when(client.invokeBinding(any(DaprProtos.InvokeBindingRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); MyObject event = new MyObject(1, "Event"); Mono result = adapter.invokeBinding("BindingName", "MyOperation", event, TypeRef.get(String.class)); settableFuture.set(responseBuilder.build()); @@ -304,10 +304,10 @@ public void invokeBindingObjectNoHotMono() { MockCallback callback = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFuture, callback, directExecutor()); when(client.invokeBinding(any(DaprProtos.InvokeBindingRequest.class))) - .thenAnswer(c -> { - settableFuture.set(Empty.newBuilder().build()); - return settableFuture; - }); + .thenAnswer(c -> { + settableFuture.set(Empty.newBuilder().build()); + return settableFuture; + }); MyObject event = new MyObject(1, "Event"); adapter.invokeBinding("BindingName", "MyOperation", event); // Do not call block() on mono above, so nothing should happen. @@ -317,7 +317,7 @@ public void invokeBindingObjectNoHotMono() { @Test(expected = RuntimeException.class) public void invokeServiceVoidExceptionThrownTest() { when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenThrow(RuntimeException.class); + .thenThrow(RuntimeException.class); Mono result = adapter.invokeService("appId", "method", "request", HttpExtension.NONE); result.block(); } @@ -326,7 +326,7 @@ public void invokeServiceVoidExceptionThrownTest() { public void invokeServiceIllegalArgumentExceptionThrownTest() { SettableFuture settableFuture = SettableFuture.create(); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); // HttpExtension cannot be null Mono result = adapter.invokeService("appId", "method", "request", null); result.block(); @@ -335,7 +335,7 @@ public void invokeServiceIllegalArgumentExceptionThrownTest() { @Test(expected = RuntimeException.class) public void invokeServiceEmptyRequestVoidExceptionThrownTest() { when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenThrow(RuntimeException.class); + .thenThrow(RuntimeException.class); Mono result = adapter.invokeService("appId", "method", HttpExtension.NONE, (Map)null); result.block(); } @@ -348,7 +348,7 @@ public void invokeServiceVoidCallbackExceptionThrownTest() { addCallback(settableFuture, callback, directExecutor()); settableFuture.setException(ex); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeService("appId", "method", "request", HttpExtension.NONE); result.block(); } @@ -361,7 +361,7 @@ public void invokeServiceVoidTest() throws Exception { .setData(getAny("Value")).build()); addCallback(settableFuture, callback, directExecutor()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeService("appId", "method", "request", HttpExtension.NONE); settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny("Value")).build()); result.block(); @@ -376,7 +376,7 @@ public void invokeServiceVoidObjectTest() throws Exception { .setData(getAny("Value")).build()); addCallback(settableFuture, callback, directExecutor()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); MyObject request = new MyObject(1, "Event"); Mono result = adapter.invokeService("appId", "method", request, HttpExtension.NONE); settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny("Value")).build()); @@ -387,7 +387,7 @@ public void invokeServiceVoidObjectTest() throws Exception { @Test(expected = RuntimeException.class) public void invokeServiceExceptionThrownTest() { when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenThrow(RuntimeException.class); + .thenThrow(RuntimeException.class); Mono result = adapter.invokeService("appId", "method", "request", HttpExtension.NONE, null, String.class); result.block(); } @@ -395,7 +395,7 @@ public void invokeServiceExceptionThrownTest() { @Test(expected = RuntimeException.class) public void invokeServiceNoRequestClassExceptionThrownTest() { when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenThrow(RuntimeException.class); + .thenThrow(RuntimeException.class); Mono result = adapter.invokeService("appId", "method", HttpExtension.NONE, (Map)null, String.class); result.block(); } @@ -403,7 +403,7 @@ public void invokeServiceNoRequestClassExceptionThrownTest() { @Test(expected = RuntimeException.class) public void invokeServiceNoRequestTypeRefExceptionThrownTest() { when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenThrow(RuntimeException.class); + .thenThrow(RuntimeException.class); Mono result = adapter.invokeService("appId", "method", HttpExtension.NONE, (Map)null, TypeRef.STRING); result.block(); } @@ -415,7 +415,7 @@ public void invokeServiceCallbackExceptionThrownTest() { MockCallback callback = new MockCallback<>(ex); addCallback(settableFuture, callback, directExecutor()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeService("appId", "method", "request", HttpExtension.NONE, null, String.class); settableFuture.setException(ex); result.block(); @@ -427,17 +427,17 @@ public void invokeServiceWithHttpExtensionTest() throws IOException { put("test", "1"); }}); CommonProtos.InvokeRequest message = CommonProtos.InvokeRequest.newBuilder() - .setMethod("method") - .setData(getAny("request")) - .setContentType("application/json") - .setHttpExtension(CommonProtos.HTTPExtension.newBuilder() - .setVerb(CommonProtos.HTTPExtension.Verb.GET) - .putQuerystring("test", "1").build()) - .build(); + .setMethod("method") + .setData(getAny("request")) + .setContentType("application/json") + .setHttpExtension(CommonProtos.HTTPExtension.newBuilder() + .setVerb(CommonProtos.HTTPExtension.Verb.GET) + .putQuerystring("test", "1").build()) + .build(); DaprProtos.InvokeServiceRequest request = DaprProtos.InvokeServiceRequest.newBuilder() - .setId("appId") - .setMessage(message) - .build(); + .setId("appId") + .setMessage(message) + .build(); String expected = "Value"; SettableFuture settableFuture = SettableFuture.create(); MockCallback callback = new MockCallback<>(CommonProtos.InvokeResponse.newBuilder() @@ -445,7 +445,7 @@ public void invokeServiceWithHttpExtensionTest() throws IOException { addCallback(settableFuture, callback, directExecutor()); settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny(expected)).build()); when(client.invokeService(eq(request))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeService("appId", "method", "request", httpExtension, null, String.class); String strOutput = result.block(); assertEquals(expected, strOutput); @@ -460,7 +460,7 @@ public void invokeServiceTest() throws Exception { addCallback(settableFuture, callback, directExecutor()); settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny(expected)).build()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeService("appId", "method", "request", HttpExtension.NONE, null, String.class); String strOutput = result.block(); assertEquals(expected, strOutput); @@ -475,7 +475,7 @@ public void invokeServiceObjectTest() throws Exception { addCallback(settableFuture, callback, directExecutor()); settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny(object)).build()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeService("appId", "method", "request", HttpExtension.NONE, null, MyObject.class); MyObject resultObject = result.block(); assertEquals(object.id, resultObject.id); @@ -485,7 +485,7 @@ public void invokeServiceObjectTest() throws Exception { @Test(expected = RuntimeException.class) public void invokeServiceNoRequestBodyExceptionThrownTest() { when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenThrow(RuntimeException.class); + .thenThrow(RuntimeException.class); Mono result = adapter.invokeService("appId", "method", (Object)null, HttpExtension.NONE, String.class); result.block(); } @@ -497,7 +497,7 @@ public void invokeServiceNoRequestCallbackExceptionThrownTest() { MockCallback callback = new MockCallback<>(ex); addCallback(settableFuture, callback, directExecutor()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeService("appId", "method", (Object)null, HttpExtension.NONE, String.class); settableFuture.setException(ex); result.block(); @@ -513,7 +513,7 @@ public void invokeServiceNoRequestBodyTest() throws Exception { addCallback(settableFuture, callback, directExecutor()); settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny(expected)).build()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeService("appId", "method", (Object)null, HttpExtension.NONE, String.class); String strOutput = result.block(); assertEquals(expected, strOutput); @@ -529,7 +529,7 @@ public void invokeServiceNoRequestBodyObjectTest() throws Exception { addCallback(settableFuture, callback, directExecutor()); settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny(object)).build()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeService("appId", "method", (Object)null, HttpExtension.NONE, MyObject.class); MyObject resultObject = result.block(); assertEquals(object.id, resultObject.id); @@ -539,7 +539,7 @@ public void invokeServiceNoRequestBodyObjectTest() throws Exception { @Test(expected = RuntimeException.class) public void invokeServiceByteRequestExceptionThrownTest() throws IOException { when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenThrow(RuntimeException.class); + .thenThrow(RuntimeException.class); String request = "Request"; byte[] byteRequest = serializer.serialize(request); Mono result = adapter.invokeService("appId", "method", byteRequest, HttpExtension.NONE, byte[].class); @@ -553,11 +553,11 @@ public void invokeServiceByteRequestCallbackExceptionThrownTest() throws IOExcep MockCallback callback = new MockCallback<>(ex); addCallback(settableFuture, callback, directExecutor()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); String request = "Request"; byte[] byteRequest = serializer.serialize(request); Mono result = - adapter.invokeService("appId", "method", byteRequest, HttpExtension.NONE,(HashMap) null); + adapter.invokeService("appId", "method", byteRequest, HttpExtension.NONE,(HashMap) null); settableFuture.setException(ex); result.block(); } @@ -571,11 +571,11 @@ public void invokeByteRequestServiceTest() throws Exception { addCallback(settableFuture, callback, directExecutor()); settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny(expected)).build()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); String request = "Request"; byte[] byteRequest = serializer.serialize(request); Mono result = adapter.invokeService( - "appId", "method", byteRequest, HttpExtension.NONE, (HashMap) null); + "appId", "method", byteRequest, HttpExtension.NONE, (HashMap) null); byte[] byteOutput = result.block(); String strOutput = serializer.deserialize(byteOutput, String.class); assertEquals(expected, strOutput); @@ -590,7 +590,7 @@ public void invokeServiceByteRequestObjectTest() throws Exception { addCallback(settableFuture, callback, directExecutor()); settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny(resultObj)).build()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); String request = "Request"; byte[] byteRequest = serializer.serialize(request); Mono result = adapter.invokeService("appId", "method", byteRequest, HttpExtension.NONE, byte[].class); @@ -601,7 +601,7 @@ public void invokeServiceByteRequestObjectTest() throws Exception { @Test(expected = RuntimeException.class) public void invokeServiceNoRequestNoClassBodyExceptionThrownTest() { when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenThrow(RuntimeException.class); + .thenThrow(RuntimeException.class); Mono result = adapter.invokeService("appId", "method", (Object)null, HttpExtension.NONE); result.block(); } @@ -613,7 +613,7 @@ public void invokeServiceNoRequestNoClassCallbackExceptionThrownTest() { MockCallback callback = new MockCallback<>(ex); addCallback(settableFuture, callback, directExecutor()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeService("appId", "method", (Object)null, HttpExtension.NONE); settableFuture.setException(ex); result.block(); @@ -627,7 +627,7 @@ public void invokeServiceNoRequestNoClassBodyTest() throws Exception { .setData(getAny(expected)).build()); addCallback(settableFuture, callback, directExecutor()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeService("appId", "method", (Object)null, HttpExtension.NONE); settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny(expected)).build()); result.block(); @@ -642,10 +642,10 @@ public void invokeServiceNoRequestNoHotMono() throws Exception { .setData(getAny(expected)).build()); addCallback(settableFuture, callback, directExecutor()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenAnswer(c -> { - settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny(expected)).build()); - return settableFuture; - }); + .thenAnswer(c -> { + settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny(expected)).build()); + return settableFuture; + }); adapter.invokeService("appId", "method", (Object)null, HttpExtension.NONE); // Do not call block() on mono above, so nothing should happen. assertFalse(callback.wasCalled); @@ -661,7 +661,7 @@ public void invokeServiceNoRequestNoClassBodyObjectTest() throws Exception { addCallback(settableFuture, callback, directExecutor()); settableFuture.set(CommonProtos.InvokeResponse.newBuilder().setData(getAny(resultObj)).build()); when(client.invokeService(any(DaprProtos.InvokeServiceRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono result = adapter.invokeService("appId", "method", (Object)null, HttpExtension.NONE); result.block(); assertTrue(callback.wasCalled); @@ -701,10 +701,10 @@ public void getStateCallbackExceptionThrownTest() { SettableFuture settableFuture = SettableFuture.create(); RuntimeException ex = new RuntimeException("An Exception"); MockCallback callback = - new MockCallback<>(ex); + new MockCallback<>(ex); addCallback(settableFuture, callback, directExecutor()); when(client.getState(any(io.dapr.v1.DaprProtos.GetStateRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); State key = buildStateKey(null, "Key1", "ETag1", null); Mono> result = adapter.getState(STATE_STORE_NAME, key, String.class); settableFuture.setException(ex); @@ -716,13 +716,13 @@ public void getStateStringValueNoOptionsTest() throws IOException { String etag = "ETag1"; String key = "key1"; String expectedValue = "Expected state"; - State expectedState = buildStateKey(expectedValue, key, etag, new HashMap<>(), null); + State expectedState = buildStateKey(expectedValue, key, etag, null); DaprProtos.GetStateResponse responseEnvelope = buildGetStateResponse(expectedValue, etag); SettableFuture settableFuture = SettableFuture.create(); MockCallback callback = new MockCallback<>(responseEnvelope); addCallback(settableFuture, callback, directExecutor()); when(client.getState(any(io.dapr.v1.DaprProtos.GetStateRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); State keyRequest = buildStateKey(null, key, etag, null); Mono> result = adapter.getState(STATE_STORE_NAME, keyRequest, String.class); settableFuture.set(responseEnvelope); @@ -742,10 +742,10 @@ public void getStateStringValueNoHotMono() throws IOException { MockCallback callback = new MockCallback<>(responseEnvelope); addCallback(settableFuture, callback, directExecutor()); when(client.getState(any(io.dapr.v1.DaprProtos.GetStateRequest.class))) - .thenAnswer(c -> { - settableFuture.set(responseEnvelope); - return settableFuture; - }); + .thenAnswer(c -> { + settableFuture.set(responseEnvelope); + return settableFuture; + }); State keyRequest = buildStateKey(null, key, etag, null); adapter.getState(STATE_STORE_NAME, keyRequest, String.class); // block() on the mono above is not called, so nothing should happen. @@ -758,17 +758,17 @@ public void getStateObjectValueWithOptionsTest() throws IOException { String key = "key1"; MyObject expectedValue = new MyObject(1, "The Value"); StateOptions options = buildStateOptions(StateOptions.Consistency.STRONG, StateOptions.Concurrency.FIRST_WRITE); - State expectedState = buildStateKey(expectedValue, key, etag, new HashMap<>(), options); + State expectedState = buildStateKey(expectedValue, key, etag, options); DaprProtos.GetStateResponse responseEnvelope = DaprProtos.GetStateResponse.newBuilder() - .setData(serialize(expectedValue)) - .setEtag(etag) - .build(); + .setData(serialize(expectedValue)) + .setEtag(etag) + .build(); State keyRequest = buildStateKey(null, key, etag, options); SettableFuture settableFuture = SettableFuture.create(); MockCallback callback = new MockCallback<>(responseEnvelope); addCallback(settableFuture, callback, directExecutor()); when(client.getState(any(io.dapr.v1.DaprProtos.GetStateRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono> result = adapter.getState(STATE_STORE_NAME, keyRequest, MyObject.class); settableFuture.set(responseEnvelope); State res = result.block(); @@ -784,11 +784,11 @@ public void getStateObjectValueWithMetadataTest() throws IOException { StateOptions options = buildStateOptions(StateOptions.Consistency.STRONG, StateOptions.Concurrency.FIRST_WRITE); Map metadata = new HashMap<>(); metadata.put("key_1", "val_1"); - State expectedState = buildStateKey(expectedValue, key, etag, new HashMap<>(), options); + State expectedState = buildStateKey(expectedValue, key, etag, options); DaprProtos.GetStateResponse responseEnvelope = DaprProtos.GetStateResponse.newBuilder() - .setData(serialize(expectedValue)) - .setEtag(etag) - .build(); + .setData(serialize(expectedValue)) + .setEtag(etag) + .build(); GetStateRequestBuilder builder = new GetStateRequestBuilder(STATE_STORE_NAME, key); builder.withMetadata(metadata).withEtag(etag).withStateOptions(options); GetStateRequest request = builder.build(); @@ -796,7 +796,7 @@ public void getStateObjectValueWithMetadataTest() throws IOException { MockCallback callback = new MockCallback<>(responseEnvelope); addCallback(settableFuture, callback, directExecutor()); when(client.getState(any(io.dapr.v1.DaprProtos.GetStateRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono>> result = adapter.getState(request, TypeRef.get(MyObject.class)); settableFuture.set(responseEnvelope); Response> res = result.block(); @@ -810,17 +810,17 @@ public void getStateObjectValueWithOptionsNoConcurrencyTest() throws IOException String key = "key1"; MyObject expectedValue = new MyObject(1, "The Value"); StateOptions options = new StateOptions(null, StateOptions.Concurrency.FIRST_WRITE); - State expectedState = buildStateKey(expectedValue, key, etag, new HashMap<>(), options); + State expectedState = buildStateKey(expectedValue, key, etag, options); DaprProtos.GetStateResponse responseEnvelope = DaprProtos.GetStateResponse.newBuilder() - .setData(serialize(expectedValue)) - .setEtag(etag) - .build(); + .setData(serialize(expectedValue)) + .setEtag(etag) + .build(); State keyRequest = buildStateKey(null, key, etag, options); SettableFuture settableFuture = SettableFuture.create(); MockCallback callback = new MockCallback<>(responseEnvelope); addCallback(settableFuture, callback, directExecutor()); when(client.getState(any(io.dapr.v1.DaprProtos.GetStateRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); Mono> result = adapter.getState(STATE_STORE_NAME, keyRequest, MyObject.class); settableFuture.set(responseEnvelope); assertEquals(expectedState, result.block()); @@ -848,32 +848,32 @@ public void getStatesIllegalArgumentExceptionTest() { }); // negative parallelism GetStatesRequest req = new GetStatesRequestBuilder(STATE_STORE_NAME, Collections.singletonList("100")) - .withParallelism(-1) - .build(); + .withParallelism(-1) + .build(); assertThrows(IllegalArgumentException.class, () -> adapter.getStates(req, TypeRef.BOOLEAN).block()); } @Test public void getStatesString() throws IOException { DaprProtos.GetBulkStateResponse responseEnvelope = DaprProtos.GetBulkStateResponse.newBuilder() - .addItems(DaprProtos.BulkStateItem.newBuilder() - .setData(serialize("hello world")) - .setKey("100") - .setEtag("1") - .build()) - .addItems(DaprProtos.BulkStateItem.newBuilder() - .setKey("200") - .setError("not found") - .build()) - .build(); + .addItems(DaprProtos.BulkStateItem.newBuilder() + .setData(serialize("hello world")) + .setKey("100") + .setEtag("1") + .build()) + .addItems(DaprProtos.BulkStateItem.newBuilder() + .setKey("200") + .setError("not found") + .build()) + .build(); SettableFuture settableFuture = SettableFuture.create(); MockCallback callback = new MockCallback<>(responseEnvelope); addCallback(settableFuture, callback, directExecutor()); when(client.getBulkState(any(DaprProtos.GetBulkStateRequest.class))) - .thenAnswer(c -> { - settableFuture.set(responseEnvelope); - return settableFuture; - }); + .thenAnswer(c -> { + settableFuture.set(responseEnvelope); + return settableFuture; + }); List> result = adapter.getStates(STATE_STORE_NAME, Arrays.asList("100", "200"), String.class).block(); assertTrue(callback.wasCalled); @@ -891,24 +891,24 @@ public void getStatesString() throws IOException { @Test public void getStatesInteger() throws IOException { DaprProtos.GetBulkStateResponse responseEnvelope = DaprProtos.GetBulkStateResponse.newBuilder() - .addItems(DaprProtos.BulkStateItem.newBuilder() - .setData(serialize(1234)) - .setKey("100") - .setEtag("1") - .build()) - .addItems(DaprProtos.BulkStateItem.newBuilder() - .setKey("200") - .setError("not found") - .build()) - .build(); + .addItems(DaprProtos.BulkStateItem.newBuilder() + .setData(serialize(1234)) + .setKey("100") + .setEtag("1") + .build()) + .addItems(DaprProtos.BulkStateItem.newBuilder() + .setKey("200") + .setError("not found") + .build()) + .build(); SettableFuture settableFuture = SettableFuture.create(); MockCallback callback = new MockCallback<>(responseEnvelope); addCallback(settableFuture, callback, directExecutor()); when(client.getBulkState(any(DaprProtos.GetBulkStateRequest.class))) - .thenAnswer(c -> { - settableFuture.set(responseEnvelope); - return settableFuture; - }); + .thenAnswer(c -> { + settableFuture.set(responseEnvelope); + return settableFuture; + }); List> result = adapter.getStates(STATE_STORE_NAME, Arrays.asList("100", "200"), int.class).block(); assertTrue(callback.wasCalled); @@ -926,24 +926,24 @@ public void getStatesInteger() throws IOException { @Test public void getStatesBoolean() throws IOException { DaprProtos.GetBulkStateResponse responseEnvelope = DaprProtos.GetBulkStateResponse.newBuilder() - .addItems(DaprProtos.BulkStateItem.newBuilder() - .setData(serialize(true)) - .setKey("100") - .setEtag("1") - .build()) - .addItems(DaprProtos.BulkStateItem.newBuilder() - .setKey("200") - .setError("not found") - .build()) - .build(); + .addItems(DaprProtos.BulkStateItem.newBuilder() + .setData(serialize(true)) + .setKey("100") + .setEtag("1") + .build()) + .addItems(DaprProtos.BulkStateItem.newBuilder() + .setKey("200") + .setError("not found") + .build()) + .build(); SettableFuture settableFuture = SettableFuture.create(); MockCallback callback = new MockCallback<>(responseEnvelope); addCallback(settableFuture, callback, directExecutor()); when(client.getBulkState(any(DaprProtos.GetBulkStateRequest.class))) - .thenAnswer(c -> { - settableFuture.set(responseEnvelope); - return settableFuture; - }); + .thenAnswer(c -> { + settableFuture.set(responseEnvelope); + return settableFuture; + }); List> result = adapter.getStates(STATE_STORE_NAME, Arrays.asList("100", "200"), boolean.class).block(); assertTrue(callback.wasCalled); @@ -961,24 +961,24 @@ public void getStatesBoolean() throws IOException { @Test public void getStatesByteArray() throws IOException { DaprProtos.GetBulkStateResponse responseEnvelope = DaprProtos.GetBulkStateResponse.newBuilder() - .addItems(DaprProtos.BulkStateItem.newBuilder() - .setData(serialize(new byte[]{1, 2, 3})) - .setKey("100") - .setEtag("1") - .build()) - .addItems(DaprProtos.BulkStateItem.newBuilder() - .setKey("200") - .setError("not found") - .build()) - .build(); + .addItems(DaprProtos.BulkStateItem.newBuilder() + .setData(serialize(new byte[]{1, 2, 3})) + .setKey("100") + .setEtag("1") + .build()) + .addItems(DaprProtos.BulkStateItem.newBuilder() + .setKey("200") + .setError("not found") + .build()) + .build(); SettableFuture settableFuture = SettableFuture.create(); MockCallback callback = new MockCallback<>(responseEnvelope); addCallback(settableFuture, callback, directExecutor()); when(client.getBulkState(any(DaprProtos.GetBulkStateRequest.class))) - .thenAnswer(c -> { - settableFuture.set(responseEnvelope); - return settableFuture; - }); + .thenAnswer(c -> { + settableFuture.set(responseEnvelope); + return settableFuture; + }); List> result = adapter.getStates(STATE_STORE_NAME, Arrays.asList("100", "200"), byte[].class).block(); assertTrue(callback.wasCalled); @@ -997,24 +997,24 @@ public void getStatesByteArray() throws IOException { public void getStatesObject() throws IOException { MyObject object = new MyObject(1, "Event"); DaprProtos.GetBulkStateResponse responseEnvelope = DaprProtos.GetBulkStateResponse.newBuilder() - .addItems(DaprProtos.BulkStateItem.newBuilder() - .setData(serialize(object)) - .setKey("100") - .setEtag("1") - .build()) - .addItems(DaprProtos.BulkStateItem.newBuilder() - .setKey("200") - .setError("not found") - .build()) - .build(); + .addItems(DaprProtos.BulkStateItem.newBuilder() + .setData(serialize(object)) + .setKey("100") + .setEtag("1") + .build()) + .addItems(DaprProtos.BulkStateItem.newBuilder() + .setKey("200") + .setError("not found") + .build()) + .build(); SettableFuture settableFuture = SettableFuture.create(); MockCallback callback = new MockCallback<>(responseEnvelope); addCallback(settableFuture, callback, directExecutor()); when(client.getBulkState(any(DaprProtos.GetBulkStateRequest.class))) - .thenAnswer(c -> { - settableFuture.set(responseEnvelope); - return settableFuture; - }); + .thenAnswer(c -> { + settableFuture.set(responseEnvelope); + return settableFuture; + }); List> result = adapter.getStates(STATE_STORE_NAME, Arrays.asList("100", "200"), MyObject.class).block(); assertTrue(callback.wasCalled); @@ -1065,7 +1065,7 @@ public void deleteStateCallbackExcpetionThrownTest() { MockCallback callback = new MockCallback<>(ex); addCallback(settableFuture, callback, directExecutor()); when(client.deleteState(any(io.dapr.v1.DaprProtos.DeleteStateRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); State key = buildStateKey(null, "Key1", "ETag1", null); Mono result = adapter.deleteState(STATE_STORE_NAME, key.getKey(), key.getEtag(), key.getOptions()); settableFuture.setException(ex); @@ -1080,10 +1080,10 @@ public void deleteStateNoOptionsTest() { MockCallback callback = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFuture, callback, directExecutor()); when(client.deleteState(any(io.dapr.v1.DaprProtos.DeleteStateRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); State stateKey = buildStateKey(null, key, etag, null); Mono result = adapter.deleteState(STATE_STORE_NAME, stateKey.getKey(), stateKey.getEtag(), - stateKey.getOptions()); + stateKey.getOptions()); settableFuture.set(Empty.newBuilder().build()); result.block(); assertTrue(callback.wasCalled); @@ -1098,10 +1098,10 @@ public void deleteStateTest() { MockCallback callback = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFuture, callback, directExecutor()); when(client.deleteState(any(io.dapr.v1.DaprProtos.DeleteStateRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); State stateKey = buildStateKey(null, key, etag, options); Mono result = adapter.deleteState(STATE_STORE_NAME, stateKey.getKey(), stateKey.getEtag(), - stateKey.getOptions()); + stateKey.getOptions()); settableFuture.set(Empty.newBuilder().build()); result.block(); assertTrue(callback.wasCalled); @@ -1118,7 +1118,7 @@ public void deleteStateWithMetadata() { MockCallback callback = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFuture, callback, directExecutor()); when(client.deleteState(any(io.dapr.v1.DaprProtos.DeleteStateRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); DeleteStateRequestBuilder builder = new DeleteStateRequestBuilder(STATE_STORE_NAME, key); builder.withEtag(etag).withStateOptions(options).withMetadata(metadata); DeleteStateRequest request = builder.build(); @@ -1137,13 +1137,13 @@ public void deleteStateTestNoHotMono() { MockCallback callback = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFuture, callback, directExecutor()); when(client.deleteState(any(io.dapr.v1.DaprProtos.DeleteStateRequest.class))) - .thenAnswer(c -> { - settableFuture.set(Empty.newBuilder().build()); - return settableFuture; - }); + .thenAnswer(c -> { + settableFuture.set(Empty.newBuilder().build()); + return settableFuture; + }); State stateKey = buildStateKey(null, key, etag, options); Mono result = adapter.deleteState(STATE_STORE_NAME, stateKey.getKey(), stateKey.getEtag(), - stateKey.getOptions()); + stateKey.getOptions()); // Do not call result.block(), so nothing should happen. assertFalse(callback.wasCalled); } @@ -1157,10 +1157,10 @@ public void deleteStateNoConsistencyTest() { MockCallback callback = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFuture, callback, directExecutor()); when(client.deleteState(any(io.dapr.v1.DaprProtos.DeleteStateRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); State stateKey = buildStateKey(null, key, etag, options); Mono result = adapter.deleteState(STATE_STORE_NAME, stateKey.getKey(), stateKey.getEtag(), - stateKey.getOptions()); + stateKey.getOptions()); settableFuture.set(Empty.newBuilder().build()); result.block(); assertTrue(callback.wasCalled); @@ -1175,10 +1175,10 @@ public void deleteStateNoConcurrencyTest() { MockCallback callback = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFuture, callback, directExecutor()); when(client.deleteState(any(io.dapr.v1.DaprProtos.DeleteStateRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); State stateKey = buildStateKey(null, key, etag, options); Mono result = adapter.deleteState(STATE_STORE_NAME, stateKey.getKey(), stateKey.getEtag(), - stateKey.getOptions()); + stateKey.getOptions()); settableFuture.set(Empty.newBuilder().build()); result.block(); assertTrue(callback.wasCalled); @@ -1188,8 +1188,8 @@ public void deleteStateNoConcurrencyTest() { public void executeTransactionIllegalArgumentExceptionTest() { State key = buildStateKey(null, "Key1", "ETag1", null); TransactionalStateOperation upsertOperation = new TransactionalStateOperation<>( - TransactionalStateOperation.OperationType.UPSERT, - key); + TransactionalStateOperation.OperationType.UPSERT, + key); assertThrows(IllegalArgumentException.class, () -> { // empty state store name adapter.executeTransaction("", Collections.singletonList(upsertOperation)).block(); @@ -1210,15 +1210,15 @@ public void executeTransactionSerializerExceptionTest() throws IOException { StateOptions options = buildStateOptions(StateOptions.Consistency.STRONG, null); SettableFuture settableFuture = SettableFuture.create(); when(client.executeStateTransaction(any(DaprProtos.ExecuteStateTransactionRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); when(mockSerializer.serialize(any())).thenThrow(IOException.class); State stateKey = buildStateKey(data, key, etag, options); TransactionalStateOperation upsertOperation = new TransactionalStateOperation<>( - TransactionalStateOperation.OperationType.UPSERT, - stateKey); + TransactionalStateOperation.OperationType.UPSERT, + stateKey); ExecuteStateTransactionRequest request = new ExecuteStateTransactionRequestBuilder(STATE_STORE_NAME) - .withTransactionalStates(upsertOperation) - .build(); + .withTransactionalStates(upsertOperation) + .build(); Mono> result = adapter.executeTransaction(request); result.block(); } @@ -1233,20 +1233,20 @@ public void executeTransactionWithMetadataTest() { MockCallback callback = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFuture, callback, directExecutor()); when(client.executeStateTransaction(any(DaprProtos.ExecuteStateTransactionRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); State stateKey = buildStateKey(data, key, etag, options); TransactionalStateOperation upsertOperation = new TransactionalStateOperation<>( - TransactionalStateOperation.OperationType.UPSERT, - stateKey); + TransactionalStateOperation.OperationType.UPSERT, + stateKey); TransactionalStateOperation deleteOperation = new TransactionalStateOperation<>( - TransactionalStateOperation.OperationType.DELETE, - new State<>("testKey")); + TransactionalStateOperation.OperationType.DELETE, + new State<>("testKey")); Map metadata = new HashMap<>(); metadata.put("testKey", "testValue"); ExecuteStateTransactionRequest request = new ExecuteStateTransactionRequestBuilder(STATE_STORE_NAME) - .withTransactionalStates(upsertOperation, deleteOperation) - .withMetadata(metadata) - .build(); + .withTransactionalStates(upsertOperation, deleteOperation) + .withMetadata(metadata) + .build(); Mono> result = adapter.executeTransaction(request); settableFuture.set(Empty.newBuilder().build()); result.block(); @@ -1263,14 +1263,14 @@ public void executeTransactionTest() { MockCallback callback = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFuture, callback, directExecutor()); when(client.executeStateTransaction(any(DaprProtos.ExecuteStateTransactionRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); State stateKey = buildStateKey(data, key, etag, options); TransactionalStateOperation upsertOperation = new TransactionalStateOperation<>( - TransactionalStateOperation.OperationType.UPSERT, - stateKey); + TransactionalStateOperation.OperationType.UPSERT, + stateKey); TransactionalStateOperation deleteOperation = new TransactionalStateOperation<>( - TransactionalStateOperation.OperationType.DELETE, - new State<>("testKey") + TransactionalStateOperation.OperationType.DELETE, + new State<>("testKey") ); Mono result = adapter.executeTransaction(STATE_STORE_NAME, Arrays.asList(upsertOperation, deleteOperation)); settableFuture.set(Empty.newBuilder().build()); @@ -1285,11 +1285,11 @@ public void executeTransactionExceptionThrownTest() { String data = "my data"; StateOptions options = buildStateOptions(StateOptions.Consistency.STRONG, null); when(client.executeStateTransaction(any(DaprProtos.ExecuteStateTransactionRequest.class))) - .thenThrow(RuntimeException.class); + .thenThrow(RuntimeException.class); State stateKey = buildStateKey(data, key, etag, options); TransactionalStateOperation operation = new TransactionalStateOperation<>( - TransactionalStateOperation.OperationType.UPSERT, - stateKey); + TransactionalStateOperation.OperationType.UPSERT, + stateKey); Mono result = adapter.executeTransaction(STATE_STORE_NAME, Collections.singletonList(operation)); result.block(); } @@ -1305,11 +1305,11 @@ public void executeTransactionCallbackExceptionTest() { MockCallback callback = new MockCallback<>(ex); addCallback(settableFuture, callback, directExecutor()); when(client.executeStateTransaction(any(DaprProtos.ExecuteStateTransactionRequest.class))) - .thenReturn(settableFuture); + .thenReturn(settableFuture); State stateKey = buildStateKey(data, key, etag, options); TransactionalStateOperation operation = new TransactionalStateOperation<>( - TransactionalStateOperation.OperationType.UPSERT, - stateKey); + TransactionalStateOperation.OperationType.UPSERT, + stateKey); Mono result = adapter.executeTransaction(STATE_STORE_NAME, Collections.singletonList(operation)); settableFuture.setException(ex); result.block(); @@ -1453,10 +1453,6 @@ private State buildStateKey(T value, String key, String etag, StateOption return new State<>(value, key, etag, options); } - private State buildStateKey(T value, String key, String etag, Map metadata, StateOptions options) { - return new State<>(value, key, etag, metadata, options); - } - /** * The purpose of this test is to show that it doesn't matter when the client is called, the actual coll to DAPR * will be done when the output Mono response call the Mono.block method. @@ -1484,7 +1480,7 @@ public void getStateDeleteStateThenBlockDeleteThenBlockGet() throws Exception { String expectedValue1 = "Expected state 1"; String key2 = "key2"; String expectedValue2 = "Expected state 2"; - State expectedState1 = buildStateKey(expectedValue1, key1, etag, new HashMap<>(), null); + State expectedState1 = buildStateKey(expectedValue1, key1, etag, null); Map> futuresMap = new HashMap<>(); futuresMap.put(key1, buildFutureGetStateEnvelop(expectedValue1, etag)); futuresMap.put(key2, buildFutureGetStateEnvelop(expectedValue2, etag)); @@ -1499,9 +1495,9 @@ public void getStateDeleteStateThenBlockDeleteThenBlockGet() throws Exception { MockCallback callbackDelete = new MockCallback<>(Empty.newBuilder().build()); addCallback(settableFutureDelete, callbackDelete, directExecutor()); when(client.deleteState(any(io.dapr.v1.DaprProtos.DeleteStateRequest.class))) - .thenReturn(settableFutureDelete); + .thenReturn(settableFutureDelete); Mono resultDelete = adapter.deleteState(STATE_STORE_NAME, keyRequest2.getKey(), keyRequest2.getEtag(), - keyRequest2.getOptions()); + keyRequest2.getOptions()); settableFutureDelete.set(Empty.newBuilder().build()); resultDelete.block(); assertTrue(callbackDelete.wasCalled); @@ -1523,13 +1519,13 @@ public void getSecrets() { settableFuture.set(responseEnvelope); when(client.getSecret(any(io.dapr.v1.DaprProtos.GetSecretRequest.class))) - .thenAnswer(context -> { - io.dapr.v1.DaprProtos.GetSecretRequest req = context.getArgument(0); - assertEquals("key", req.getKey()); - assertEquals(SECRET_STORE_NAME, req.getStoreName()); - assertEquals(0, req.getMetadataCount()); - return settableFuture; - }); + .thenAnswer(context -> { + io.dapr.v1.DaprProtos.GetSecretRequest req = context.getArgument(0); + assertEquals("key", req.getKey()); + assertEquals(SECRET_STORE_NAME, req.getStoreName()); + assertEquals(0, req.getMetadataCount()); + return settableFuture; + }); Map result = adapter.getSecret(SECRET_STORE_NAME, "key").block(); @@ -1546,13 +1542,13 @@ public void getSecretsEmptyResponse() { settableFuture.set(responseEnvelope); when(client.getSecret(any(io.dapr.v1.DaprProtos.GetSecretRequest.class))) - .thenAnswer(context -> { - io.dapr.v1.DaprProtos.GetSecretRequest req = context.getArgument(0); - assertEquals("key", req.getKey()); - assertEquals(SECRET_STORE_NAME, req.getStoreName()); - assertEquals(0, req.getMetadataCount()); - return settableFuture; - }); + .thenAnswer(context -> { + io.dapr.v1.DaprProtos.GetSecretRequest req = context.getArgument(0); + assertEquals("key", req.getKey()); + assertEquals(SECRET_STORE_NAME, req.getStoreName()); + assertEquals(0, req.getMetadataCount()); + return settableFuture; + }); Map result = adapter.getSecret(SECRET_STORE_NAME, "key").block(); @@ -1567,13 +1563,13 @@ public void getSecretsException() { settableFuture.setException(new RuntimeException()); when(client.getSecret(any(io.dapr.v1.DaprProtos.GetSecretRequest.class))) - .thenAnswer(context -> { - io.dapr.v1.DaprProtos.GetSecretRequest req = context.getArgument(0); - assertEquals("key", req.getKey()); - assertEquals(SECRET_STORE_NAME, req.getStoreName()); - assertEquals(0, req.getMetadataCount()); - return settableFuture; - }); + .thenAnswer(context -> { + io.dapr.v1.DaprProtos.GetSecretRequest req = context.getArgument(0); + assertEquals("key", req.getKey()); + assertEquals(SECRET_STORE_NAME, req.getStoreName()); + assertEquals(0, req.getMetadataCount()); + return settableFuture; + }); assertThrows(RuntimeException.class, () -> adapter.getSecret(SECRET_STORE_NAME, "key").block()); } @@ -1609,18 +1605,18 @@ public void getSecretsWithMetadata() { settableFuture.set(responseEnvelope); when(client.getSecret(any(io.dapr.v1.DaprProtos.GetSecretRequest.class))) - .thenAnswer(context -> { - io.dapr.v1.DaprProtos.GetSecretRequest req = context.getArgument(0); - assertEquals("key", req.getKey()); - assertEquals(SECRET_STORE_NAME, req.getStoreName()); - assertEquals("metavalue", req.getMetadataMap().get("metakey")); - return settableFuture; - }); + .thenAnswer(context -> { + io.dapr.v1.DaprProtos.GetSecretRequest req = context.getArgument(0); + assertEquals("key", req.getKey()); + assertEquals(SECRET_STORE_NAME, req.getStoreName()); + assertEquals("metavalue", req.getMetadataMap().get("metakey")); + return settableFuture; + }); Map result = adapter.getSecret( - SECRET_STORE_NAME, - "key", - Collections.singletonMap("metakey", "metavalue")).block(); + SECRET_STORE_NAME, + "key", + Collections.singletonMap("metakey", "metavalue")).block(); assertEquals(1, result.size()); assertEquals(expectedValue, result.get(expectedKey)); @@ -1682,15 +1678,15 @@ private SettableFuture buildFutureGetStateEnvel private DaprProtos.GetStateResponse buildGetStateResponse(T value, String etag) throws IOException { return DaprProtos.GetStateResponse.newBuilder() - .setData(serialize(value)) - .setEtag(etag) - .build(); + .setData(serialize(value)) + .setEtag(etag) + .build(); } private DaprProtos.GetSecretResponse buildGetSecretResponse(String key, String value) { return DaprProtos.GetSecretResponse.newBuilder() - .putAllData(Collections.singletonMap(key, value)) - .build(); + .putAllData(Collections.singletonMap(key, value)) + .build(); } private DaprProtos.GetSecretResponse buildGetSecretResponse() { From 2c54187fc9df843dab61b783410a41aa34257019 Mon Sep 17 00:00:00 2001 From: Sky Ao Date: Tue, 29 Dec 2020 21:45:20 +0800 Subject: [PATCH 3/6] fix compile error after merge --- .../io/dapr/client/domain/GetBulkStateRequestBuilder.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sdk/src/main/java/io/dapr/client/domain/GetBulkStateRequestBuilder.java b/sdk/src/main/java/io/dapr/client/domain/GetBulkStateRequestBuilder.java index 86501540be..28b5100d7e 100644 --- a/sdk/src/main/java/io/dapr/client/domain/GetBulkStateRequestBuilder.java +++ b/sdk/src/main/java/io/dapr/client/domain/GetBulkStateRequestBuilder.java @@ -37,12 +37,12 @@ public GetBulkStateRequestBuilder(String storeName, String... keys) { this.keys = keys == null ? null : Collections.unmodifiableList(Arrays.asList(keys)); } - public GetStatesRequestBuilder withMetadata(Map metadata) { + public GetBulkStateRequestBuilder withMetadata(Map metadata) { this.metadata = metadata == null ? null : Collections.unmodifiableMap(metadata); return this; } - public GetStatesRequestBuilder withParallelism(int parallelism) { + public GetBulkStateRequestBuilder withParallelism(int parallelism) { this.parallelism = parallelism; return this; } From e221b4ef5c19405484b1040f3a1e46c3085fdca6 Mon Sep 17 00:00:00 2001 From: Sky Ao Date: Tue, 29 Dec 2020 22:06:32 +0800 Subject: [PATCH 4/6] update unit test for metadata in getBulkState() --- .../java/io/dapr/client/DaprClientGrpcTest.java | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java b/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java index e1ed48e9c4..f082e606f3 100644 --- a/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java +++ b/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java @@ -958,10 +958,13 @@ public void getStatesIllegalArgumentExceptionTest() { @Test public void getStatesString() throws IOException { + Map metadata = new HashMap<>(); + metadata.put("meta1", "value1"); DaprProtos.GetBulkStateResponse responseEnvelope = DaprProtos.GetBulkStateResponse.newBuilder() .addItems(DaprProtos.BulkStateItem.newBuilder() .setData(serialize("hello world")) .setKey("100") + .putAllMetadata(metadata) .setEtag("1") .build()) .addItems(DaprProtos.BulkStateItem.newBuilder() @@ -983,6 +986,7 @@ public void getStatesString() throws IOException { assertEquals(2, result.size()); assertEquals("100", result.stream().findFirst().get().getKey()); assertEquals("hello world", result.stream().findFirst().get().getValue()); + assertEquals(metadata, result.stream().findFirst().get().getMetadata()); assertEquals("1", result.stream().findFirst().get().getEtag()); assertNull(result.stream().findFirst().get().getError()); assertEquals("200", result.stream().skip(1).findFirst().get().getKey()); @@ -993,10 +997,13 @@ public void getStatesString() throws IOException { @Test public void getStatesInteger() throws IOException { + Map metadata = new HashMap<>(); + metadata.put("meta1", "value1"); DaprProtos.GetBulkStateResponse responseEnvelope = DaprProtos.GetBulkStateResponse.newBuilder() .addItems(DaprProtos.BulkStateItem.newBuilder() .setData(serialize(1234)) .setKey("100") + .putAllMetadata(metadata) .setEtag("1") .build()) .addItems(DaprProtos.BulkStateItem.newBuilder() @@ -1018,6 +1025,7 @@ public void getStatesInteger() throws IOException { assertEquals(2, result.size()); assertEquals("100", result.stream().findFirst().get().getKey()); assertEquals(1234, (int)result.stream().findFirst().get().getValue()); + assertEquals(metadata, result.stream().findFirst().get().getMetadata()); assertEquals("1", result.stream().findFirst().get().getEtag()); assertNull(result.stream().findFirst().get().getError()); assertEquals("200", result.stream().skip(1).findFirst().get().getKey()); @@ -1028,10 +1036,13 @@ public void getStatesInteger() throws IOException { @Test public void getStatesBoolean() throws IOException { + Map metadata = new HashMap<>(); + metadata.put("meta1", "value1"); DaprProtos.GetBulkStateResponse responseEnvelope = DaprProtos.GetBulkStateResponse.newBuilder() .addItems(DaprProtos.BulkStateItem.newBuilder() .setData(serialize(true)) .setKey("100") + .putAllMetadata(metadata) .setEtag("1") .build()) .addItems(DaprProtos.BulkStateItem.newBuilder() @@ -1053,6 +1064,7 @@ public void getStatesBoolean() throws IOException { assertEquals(2, result.size()); assertEquals("100", result.stream().findFirst().get().getKey()); assertEquals(true, result.stream().findFirst().get().getValue()); + assertEquals(metadata, result.stream().findFirst().get().getMetadata()); assertEquals("1", result.stream().findFirst().get().getEtag()); assertNull(result.stream().findFirst().get().getError()); assertEquals("200", result.stream().skip(1).findFirst().get().getKey()); @@ -1063,10 +1075,12 @@ public void getStatesBoolean() throws IOException { @Test public void getStatesByteArray() throws IOException { + Map metadata = new HashMap<>(); DaprProtos.GetBulkStateResponse responseEnvelope = DaprProtos.GetBulkStateResponse.newBuilder() .addItems(DaprProtos.BulkStateItem.newBuilder() .setData(serialize(new byte[]{1, 2, 3})) .setKey("100") + .putAllMetadata(metadata) .setEtag("1") .build()) .addItems(DaprProtos.BulkStateItem.newBuilder() @@ -1088,6 +1102,7 @@ public void getStatesByteArray() throws IOException { assertEquals(2, result.size()); assertEquals("100", result.stream().findFirst().get().getKey()); assertArrayEquals(new byte[]{1, 2, 3}, result.stream().findFirst().get().getValue()); + assertEquals(0, result.stream().findFirst().get().getMetadata().size()); assertEquals("1", result.stream().findFirst().get().getEtag()); assertNull(result.stream().findFirst().get().getError()); assertEquals("200", result.stream().skip(1).findFirst().get().getKey()); @@ -1124,6 +1139,7 @@ public void getStatesObject() throws IOException { assertEquals(2, result.size()); assertEquals("100", result.stream().findFirst().get().getKey()); assertEquals(object, result.stream().findFirst().get().getValue()); + assertEquals(0, result.stream().findFirst().get().getMetadata().size()); assertEquals("1", result.stream().findFirst().get().getEtag()); assertNull(result.stream().findFirst().get().getError()); assertEquals("200", result.stream().skip(1).findFirst().get().getKey()); From ab95648d85f0e8866d789e5e30c4251a5edd0268 Mon Sep 17 00:00:00 2001 From: Sky Ao Date: Tue, 29 Dec 2020 22:36:49 +0800 Subject: [PATCH 5/6] re-trigger build again --- sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java | 1 + 1 file changed, 1 insertion(+) diff --git a/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java b/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java index f082e606f3..ceceacb420 100644 --- a/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java +++ b/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java @@ -960,6 +960,7 @@ public void getStatesIllegalArgumentExceptionTest() { public void getStatesString() throws IOException { Map metadata = new HashMap<>(); metadata.put("meta1", "value1"); + metadata.put("meta2", "value2"); DaprProtos.GetBulkStateResponse responseEnvelope = DaprProtos.GetBulkStateResponse.newBuilder() .addItems(DaprProtos.BulkStateItem.newBuilder() .setData(serialize("hello world")) From 5ffbb63953fc51ce0045f2896ecc1ac75bbb0089 Mon Sep 17 00:00:00 2001 From: Sky Ao Date: Tue, 29 Dec 2020 22:56:56 +0800 Subject: [PATCH 6/6] add test for withMetadata() --- sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java | 1 + 1 file changed, 1 insertion(+) diff --git a/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java b/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java index ceceacb420..0a341480db 100644 --- a/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java +++ b/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java @@ -951,6 +951,7 @@ public void getStatesIllegalArgumentExceptionTest() { }); // negative parallelism GetBulkStateRequest req = new GetBulkStateRequestBuilder(STATE_STORE_NAME, Collections.singletonList("100")) + .withMetadata(new HashMap<>()) .withParallelism(-1) .build(); assertThrows(IllegalArgumentException.class, () -> adapter.getBulkState(req, TypeRef.BOOLEAN).block());