This is an administrative operation that only needs to be done once for the lifetime of the task hub.
- * - * @param recreateIfExiststrue to delete any existing task hub first; false to make this
- * operation a no-op if the task hub data store already exists. Note that deleting a task
- * hub will result in permanent data loss. Use this operation with care.
- */
- public abstract void createTaskHub(boolean recreateIfExists);
-
- /**
- * Permanently deletes the target task hub data store and any orchestration data it may contain.
- *
- * This is an administrative operation that is irreversible. It should be used with great care.
- */ - public abstract void deleteTaskHub(); - /** * Purges orchestration instance metadata from the durable store. * diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcClient.java b/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcClient.java index 881b9e958..e66e6f308 100644 --- a/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcClient.java +++ b/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcClient.java @@ -32,13 +32,12 @@ import io.opentelemetry.api.trace.Tracer; import javax.annotation.Nullable; + import java.io.FileInputStream; import java.io.IOException; import java.io.InputStream; import java.time.Duration; import java.time.Instant; -import java.util.ArrayList; -import java.util.List; import java.util.Optional; import java.util.UUID; import java.util.concurrent.TimeUnit; @@ -307,49 +306,6 @@ public void terminate(String instanceId, @Nullable Object output) { this.sidecarClient.terminateInstance(builder.build()); } - @Override - public OrchestrationStatusQueryResult queryInstances(OrchestrationStatusQuery query) { - OrchestratorService.InstanceQuery.Builder instanceQueryBuilder = OrchestratorService.InstanceQuery.newBuilder(); - Optional.ofNullable(query.getCreatedTimeFrom()).ifPresent(createdTimeFrom -> - instanceQueryBuilder.setCreatedTimeFrom(DataConverter.getTimestampFromInstant(createdTimeFrom))); - Optional.ofNullable(query.getCreatedTimeTo()).ifPresent(createdTimeTo -> - instanceQueryBuilder.setCreatedTimeTo(DataConverter.getTimestampFromInstant(createdTimeTo))); - Optional.ofNullable(query.getContinuationToken()).ifPresent(token -> - instanceQueryBuilder.setContinuationToken(StringValue.of(token))); - Optional.ofNullable(query.getInstanceIdPrefix()).ifPresent(prefix -> - instanceQueryBuilder.setInstanceIdPrefix(StringValue.of(prefix))); - instanceQueryBuilder.setFetchInputsAndOutputs(query.isFetchInputsAndOutputs()); - instanceQueryBuilder.setMaxInstanceCount(query.getMaxInstanceCount()); - query.getRuntimeStatusList().forEach(runtimeStatus -> - Optional.ofNullable(runtimeStatus).ifPresent(status -> - instanceQueryBuilder.addRuntimeStatus(OrchestrationRuntimeStatus.toProtobuf(status)))); - query.getTaskHubNames().forEach(taskHubName -> Optional.ofNullable(taskHubName).ifPresent(name -> - instanceQueryBuilder.addTaskHubNames(StringValue.of(name)))); - OrchestratorService.QueryInstancesResponse queryInstancesResponse = this.sidecarClient - .queryInstances(OrchestratorService.QueryInstancesRequest.newBuilder().setQuery(instanceQueryBuilder).build()); - return toQueryResult(queryInstancesResponse, query.isFetchInputsAndOutputs()); - } - - private OrchestrationStatusQueryResult toQueryResult( - OrchestratorService.QueryInstancesResponse queryInstancesResponse, boolean fetchInputsAndOutputs) { - ListRequests may return fewer records than the specified page size, even if there are more records. - * Always check the continuation token to determine whether there are more records.
- * - * @param maxInstanceCount the maximum number of orchestration metadata records to return - * @return this query object - */ - public OrchestrationStatusQuery setMaxInstanceCount(int maxInstanceCount) { - this.maxInstanceCount = maxInstanceCount; - return this; - } - - /** - * Include orchestration metadata records that have a matching task hub name. - * - * @param taskHubNames the task hub name to match or {@code null} to disable this filter - * @return this query object - */ - public OrchestrationStatusQuery setTaskHubNames(@Nullable ListThis should always be the continuation token value from the previous query's - * {@link OrchestrationStatusQueryResult} result.
- * - * @param continuationToken the continuation token from the previous query - * @return this query object - */ - public OrchestrationStatusQuery setContinuationToken(@Nullable String continuationToken) { - this.continuationToken = continuationToken; - return this; - } - - /** - * Include orchestration metadata records with the specified instance ID prefix. - * - *For example, if there are three orchestration instances in the metadata store with IDs "Foo", "Bar", and "Baz", - * specifying a prefix value of "B" will exclude "Foo" since its ID doesn't start with "B".
- * - * @param instanceIdPrefix the instance ID prefix filter value - * @return this query object - */ - public OrchestrationStatusQuery setInstanceIdPrefix(@Nullable String instanceIdPrefix) { - this.instanceIdPrefix = instanceIdPrefix; - return this; - } - - /** - * Sets whether to fetch orchestration inputs, outputs, and custom status values. The default value is {@code false}. - * - * @param fetchInputsAndOutputs {@code true} to fetch orchestration inputs, outputs, and custom status values, - * otherwise {@code false} - * @return this query object - */ - public OrchestrationStatusQuery setFetchInputsAndOutputs(boolean fetchInputsAndOutputs) { - this.fetchInputsAndOutputs = fetchInputsAndOutputs; - return this; - } - - /** - * Gets the configured runtime status filter or {@code null} if none was configured. - * - * @return the configured runtime status filter as a list of values or {@code null} if none was configured - */ - public ListOrchestration metadata can be queried with filters using the {@link DurableTaskClient#queryInstances} method.
- */ -public final class OrchestrationStatusQueryResult { - private final ListNote that a non-null value does not always mean that there are more metadata records that can be returned by a - * query.
- * - * @return the continuation token to use with the next query or {@code null} if no more metadata records are found. - */ - public String getContinuationToken() { - return this.continuationToken; - } -} diff --git a/durabletask-client/src/test/java/io/dapr/durabletask/DurableTaskClientIT.java b/durabletask-client/src/test/java/io/dapr/durabletask/DurableTaskClientIT.java index abf146c7c..dcd43dc49 100644 --- a/durabletask-client/src/test/java/io/dapr/durabletask/DurableTaskClientIT.java +++ b/durabletask-client/src/test/java/io/dapr/durabletask/DurableTaskClientIT.java @@ -1100,188 +1100,6 @@ void clearCustomStatus() throws TimeoutException { } } - // due to clock drift, client/worker and sidecar time are not exactly synchronized, this test needs to accommodate for client vs backend timestamps difference - @Test - @Disabled("Test is disabled for investigation, fixing the test retry pattern exposed the failure") - void multiInstanceQuery() throws TimeoutException { - final String plusOne = "plusOne"; - final String waitForEvent = "waitForEvent"; - final DurableTaskClient client = new DurableTaskGrpcClientBuilder().build(); - DurableTaskGrpcWorker worker = this.createWorkerBuilder() - .addOrchestrator(plusOne, ctx -> { - int value = ctx.getInput(int.class); - for (int i = 0; i < 10; i++) { - value = ctx.callActivity(plusOne, value, int.class).await(); - } - ctx.complete(value); - }) - .addActivity(plusOne, ctx -> ctx.getInput(int.class) + 1) - .addOrchestrator(waitForEvent, ctx -> { - String name = ctx.getInput(String.class); - String output = ctx.waitForExternalEvent(name, String.class).await(); - ctx.complete(output); - }).buildAndStart(); - - try (worker; client) { - Instant startTime = Instant.now(); - String prefix = startTime.toString(); - - IntStream.range(0, 5).mapToObj(i -> { - String instanceId = String.format("%s.sequence.%d", prefix, i); - client.scheduleNewOrchestrationInstance(plusOne, 0, instanceId); - return instanceId; - }).collect(Collectors.toUnmodifiableList()).forEach(id -> { - try { - client.waitForInstanceCompletion(id, defaultTimeout, true); - } catch (TimeoutException e) { - e.printStackTrace(); - } - }); - - try { - Thread.sleep(2000); - } catch (InterruptedException e) { - } - - Instant sequencesFinishedTime = Instant.now(); - - IntStream.range(0, 5).mapToObj(i -> { - String instanceId = String.format("%s.waiter.%d", prefix, i); - client.scheduleNewOrchestrationInstance(waitForEvent, String.valueOf(i), instanceId); - return instanceId; - }).collect(Collectors.toUnmodifiableList()).forEach(id -> { - try { - client.waitForInstanceStart(id, defaultTimeout); - } catch (TimeoutException e) { - e.printStackTrace(); - } - }); - - // Create one query object and reuse it for multiple queries - OrchestrationStatusQuery query = new OrchestrationStatusQuery(); - OrchestrationStatusQueryResult result = null; - - // Return all instances - result = client.queryInstances(query); - assertEquals(10, result.getOrchestrationState().size()); - - // Test CreatedTimeTo filter - query.setCreatedTimeTo(startTime.minus(Duration.ofSeconds(1))); - result = client.queryInstances(query); - assertTrue(result.getOrchestrationState().isEmpty(), - "Result should be empty but found " + result.getOrchestrationState().size() + " instances: " + - "Start time: " + startTime + ", " + - result.getOrchestrationState().stream() - .map(state -> String.format("\nID: %s, Status: %s, Created: %s", - state.getInstanceId(), - state.getRuntimeStatus(), - state.getCreatedAt())) - .collect(Collectors.joining(", "))); - - query.setCreatedTimeTo(sequencesFinishedTime); - result = client.queryInstances(query); - // Verify all returned instances contain "sequence" in their IDs - assertEquals(5, result.getOrchestrationState().stream() - .filter(state -> state.getInstanceId().contains("sequence")) - .count(), - "Expected exactly 5 instances with 'sequence' in their IDs"); - - query.setCreatedTimeTo(Instant.now().plus(Duration.ofSeconds(1))); - result = client.queryInstances(query); - assertEquals(10, result.getOrchestrationState().size()); - - // Test CreatedTimeFrom filter - query.setCreatedTimeFrom(Instant.now().plus(Duration.ofSeconds(1))); - result = client.queryInstances(query); - assertTrue(result.getOrchestrationState().isEmpty()); - - query.setCreatedTimeFrom(sequencesFinishedTime.minus(Duration.ofSeconds(5))); - result = client.queryInstances(query); - assertEquals(5, result.getOrchestrationState().stream() - .filter(state -> state.getInstanceId().contains("sequence")) - .count(), - "Expected exactly 5 instances with 'sequence' in their IDs"); - - query.setCreatedTimeFrom(startTime.minus(Duration.ofSeconds(1))); - result = client.queryInstances(query); - assertEquals(10, result.getOrchestrationState().size()); - - // Test RuntimeStatus filter - HashSet