more fixes

Author: bachuv

client/src/main/java/com/microsoft/durabletask/AbstractTaskEntity.java

@@ -47,7 +47,6 @@ public final class DurableTaskGrpcWorker implements AutoCloseable {
     private final Duration maximumTimerInterval;
     private final DurableTaskGrpcWorkerVersioningOptions versioningOptions;
     private final int maxConcurrentEntityWorkItems;
-    private final int maxConcurrentActivityWorkItems;
     private final ExecutorService workItemExecutor;
 
     private final TaskHubSidecarServiceBlockingStub sidecarClient;
@@ -57,7 +56,6 @@ public final class DurableTaskGrpcWorker implements AutoCloseable {
         this.activityFactories.putAll(builder.activityFactories);
         this.entityFactories.putAll(builder.entityFactories);
         this.maxConcurrentEntityWorkItems = builder.maxConcurrentEntityWorkItems;
-        this.maxConcurrentActivityWorkItems = builder.maxConcurrentActivityWorkItems;
 
         Channel sidecarGrpcChannel;
         if (builder.channel != null) {
@@ -170,9 +168,6 @@ public void startAndBlock() {
         while (true) {
             try {
                 GetWorkItemsRequest.Builder requestBuilder = GetWorkItemsRequest.newBuilder();
-                if (this.maxConcurrentActivityWorkItems > 0) {
-                    requestBuilder.setMaxConcurrentActivityWorkItems(this.maxConcurrentActivityWorkItems);
-                }
                 if (!this.entityFactories.isEmpty()) {
                     // Signal to the sidecar that this worker can handle entity work items
                     requestBuilder.setMaxConcurrentEntityWorkItems(this.maxConcurrentEntityWorkItems);
@@ -385,54 +380,40 @@ public void startAndBlock() {
                                 spanAttributes);
                         Scope activityScope = activitySpan.makeCurrent();
 
-                        this.workItemExecutor.submit(() -> {
-                            String output = null;
-                            TaskFailureDetails failureDetails = null;
-                            Throwable activityError = null;
-                            try {
-                                output = taskActivityExecutor.execute(
-                                    activityRequest.getName(),
-                                    activityRequest.getInput().getValue(),
-                                    activityRequest.getTaskId());
-                            } catch (Throwable e) {
-                                activityError = e;
-                                failureDetails = TaskFailureDetails.newBuilder()
-                                    .setErrorType(e.getClass().getName())
-                                    .setErrorMessage(e.getMessage())
-                                    .setStackTrace(StringValue.of(FailureDetails.getFullStackTrace(e)))
-                                    .build();
-                            } finally {
-                                activityScope.close();
-                                TracingHelper.endSpan(activitySpan, activityError);
-                            }
+                        String output = null;
+                        TaskFailureDetails failureDetails = null;
+                        Throwable activityError = null;
+                        try {
+                            output = taskActivityExecutor.execute(
+                                activityRequest.getName(),
+                                activityRequest.getInput().getValue(),
+                                activityRequest.getTaskId());
+                        } catch (Throwable e) {
+                            activityError = e;
+                            failureDetails = TaskFailureDetails.newBuilder()
+                                .setErrorType(e.getClass().getName())
+                                .setErrorMessage(e.getMessage())
+                                .setStackTrace(StringValue.of(FailureDetails.getFullStackTrace(e)))
+                                .build();
+                        } finally {
+                            activityScope.close();
+                            TracingHelper.endSpan(activitySpan, activityError);
+                        }
 
-                            try {
-                                ActivityResponse.Builder responseBuilder = ActivityResponse.newBuilder()
-                                        .setInstanceId(activityInstanceId)
-                                        .setTaskId(activityRequest.getTaskId())
-                                        .setCompletionToken(workItem.getCompletionToken());
+                        ActivityResponse.Builder responseBuilder = ActivityResponse.newBuilder()
+                                .setInstanceId(activityInstanceId)
+                                .setTaskId(activityRequest.getTaskId())
+                                .setCompletionToken(workItem.getCompletionToken());
 
-                                if (output != null) {
-                                    responseBuilder.setResult(StringValue.of(output));
-                                }
+                        if (output != null) {
+                            responseBuilder.setResult(StringValue.of(output));
+                        }
 
-                                if (failureDetails != null) {
-                                    responseBuilder.setFailureDetails(failureDetails);
-                                }
+                        if (failureDetails != null) {
+                            responseBuilder.setFailureDetails(failureDetails);
+                        }
 
-                                this.sidecarClient.completeActivityTask(responseBuilder.build());
-                            } catch (Exception e) {
-                                logger.log(Level.WARNING,
-                                        String.format("Failed to complete activity '%s' for instance '%s'. Abandoning work item.",
-                                                activityRequest.getName(),
-                                                activityInstanceId),
-                                        e);
-                                this.sidecarClient.abandonTaskActivityWorkItem(
-                                        AbandonActivityTaskRequest.newBuilder()
-                                                .setCompletionToken(workItem.getCompletionToken())
-                                                .build());
-                            }
-                        });
+                        this.sidecarClient.completeActivityTask(responseBuilder.build());
                     } else if (requestType == RequestCase.ENTITYREQUEST) {
                         EntityBatchRequest entityRequest = workItem.getEntityRequest();
                         this.workItemExecutor.submit(() -> {

client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java

@@ -22,7 +22,6 @@ public final class DurableTaskGrpcWorkerBuilder {
     Duration maximumTimerInterval;
     DurableTaskGrpcWorkerVersioningOptions versioningOptions;
     int maxConcurrentEntityWorkItems = 1;
-    int maxConcurrentActivityWorkItems;
     int maxWorkItemThreads;
 
     /**
@@ -96,16 +95,16 @@ public DurableTaskGrpcWorkerBuilder addEntity(String name, TaskEntityFactory fac
     /**
      * Registers an entity type for the constructed {@link DurableTaskGrpcWorker}.
      * <p>
-     * The entity class must implement {@link ITaskEntity} and have a public no-argument constructor.
+     * The entity class must implement {@link TaskEntity} and have a public no-argument constructor.
      * A new instance of the entity is created for each operation batch using reflection.
      * <p>
      * The entity name is derived from the simple class name of the provided type.
      *
-     * @param entityClass the entity class to register; must implement {@link ITaskEntity}
+     * @param entityClass the entity class to register; must implement {@link TaskEntity}
      * @return this builder object
-     * @throws IllegalArgumentException if the class does not implement {@link ITaskEntity}
+     * @throws IllegalArgumentException if the class does not implement {@link TaskEntity}
      */
-    public DurableTaskGrpcWorkerBuilder addEntity(Class<? extends ITaskEntity> entityClass) {
+    public DurableTaskGrpcWorkerBuilder addEntity(Class<? extends TaskEntity> entityClass) {
         if (entityClass == null) {
             throw new IllegalArgumentException("entityClass must not be null.");
         }
@@ -116,21 +115,21 @@ public DurableTaskGrpcWorkerBuilder addEntity(Class<? extends ITaskEntity> entit
     /**
      * Registers an entity type with a specific name for the constructed {@link DurableTaskGrpcWorker}.
      * <p>
-     * The entity class must implement {@link ITaskEntity} and have a public no-argument constructor.
+     * The entity class must implement {@link TaskEntity} and have a public no-argument constructor.
      * A new instance of the entity is created for each operation batch using reflection.
      *
      * @param name        the name of the entity type
-     * @param entityClass the entity class to register; must implement {@link ITaskEntity}
+     * @param entityClass the entity class to register; must implement {@link TaskEntity}
      * @return this builder object
-     * @throws IllegalArgumentException if the class does not implement {@link ITaskEntity}
+     * @throws IllegalArgumentException if the class does not implement {@link TaskEntity}
      */
-    public DurableTaskGrpcWorkerBuilder addEntity(String name, Class<? extends ITaskEntity> entityClass) {
+    public DurableTaskGrpcWorkerBuilder addEntity(String name, Class<? extends TaskEntity> entityClass) {
         if (entityClass == null) {
             throw new IllegalArgumentException("entityClass must not be null.");
         }
-        if (!ITaskEntity.class.isAssignableFrom(entityClass)) {
+        if (!TaskEntity.class.isAssignableFrom(entityClass)) {
             throw new IllegalArgumentException(
-                    String.format("Type %s does not implement ITaskEntity.", entityClass.getName()));
+                    String.format("Type %s does not implement TaskEntity.", entityClass.getName()));
         }
         return this.addEntity(name, () -> {
             try {
@@ -152,14 +151,14 @@ public DurableTaskGrpcWorkerBuilder addEntity(String name, Class<? extends ITask
      * <p>
      * <b>Thread safety warning:</b> Because the same instance handles all operation batches,
      * the entity implementation must be thread-safe if concurrent entity work items are enabled.
-     * Implementations that extend {@link TaskEntity} store mutable state in instance fields and
+     * Implementations that extend {@link AbstractTaskEntity} store mutable state in instance fields and
      * are <b>not</b> safe for singleton registration. Use {@link #addEntity(Class)} or
      * {@link #addEntity(String, Class)} instead to create a new instance per batch.
      *
      * @param entity the entity instance to register
      * @return this builder object
      */
-    public DurableTaskGrpcWorkerBuilder addEntity(ITaskEntity entity) {
+    public DurableTaskGrpcWorkerBuilder addEntity(TaskEntity entity) {
         if (entity == null) {
             throw new IllegalArgumentException("entity must not be null.");
         }
@@ -174,15 +173,15 @@ public DurableTaskGrpcWorkerBuilder addEntity(ITaskEntity entity) {
      * <p>
      * <b>Thread safety warning:</b> Because the same instance handles all operation batches,
      * the entity implementation must be thread-safe if concurrent entity work items are enabled.
-     * Implementations that extend {@link TaskEntity} store mutable state in instance fields and
+     * Implementations that extend {@link AbstractTaskEntity} store mutable state in instance fields and
      * are <b>not</b> safe for singleton registration. Use {@link #addEntity(String, Class)} instead
      * to create a new instance per batch.
      *
      * @param name   the name of the entity type
      * @param entity the entity instance to register
      * @return this builder object
      */
-    public DurableTaskGrpcWorkerBuilder addEntity(String name, ITaskEntity entity) {
+    public DurableTaskGrpcWorkerBuilder addEntity(String name, TaskEntity entity) {
         if (entity == null) {
             throw new IllegalArgumentException("entity must not be null.");
         }
@@ -260,25 +259,7 @@ public DurableTaskGrpcWorkerBuilder maxConcurrentEntityWorkItems(int maxConcurre
     }
 
     /**
-     * Sets the maximum number of activity work items that can be processed concurrently by this worker.
-     * <p>
-     * The sidecar enforces this limit by controlling how many activity work items are dispatched
-     * to this worker at a time. The default value is 10.
-     *
-     * @param maxConcurrentActivityWorkItems the maximum number of concurrent activity work items (must be at least 1)
-     * @return this builder object
-     * @throws IllegalArgumentException if the value is less than 1
-     */
-    public DurableTaskGrpcWorkerBuilder maxConcurrentActivityWorkItems(int maxConcurrentActivityWorkItems) {
-        if (maxConcurrentActivityWorkItems < 1) {
-            throw new IllegalArgumentException("maxConcurrentActivityWorkItems must be at least 1.");
-        }
-        this.maxConcurrentActivityWorkItems = maxConcurrentActivityWorkItems;
-        return this;
-    }
-
-    /**
-     * Sets the maximum number of threads used for processing activity and entity work items.
+     * Sets the maximum number of threads used for processing entity work items.
      * <p>
      * The default value is {@value DurableTaskGrpcWorker#DEFAULT_MAX_WORK_ITEM_THREADS}.
      * Threads are created on demand and idle threads are reclaimed after 60 seconds.

client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorkerBuilder.java

@@ -106,7 +106,7 @@ public static byte[] loadAndRun(byte[] entityRequestBytes, TaskEntityFactory ent
      * @throws IllegalArgumentException if either parameter is {@code null} or if {@code base64EncodedEntityRequest}
      *                                  is not valid base64-encoded protobuf
      */
-    public static String loadAndRun(String base64EncodedEntityRequest, ITaskEntity entity) {
+    public static String loadAndRun(String base64EncodedEntityRequest, TaskEntity entity) {
         if (entity == null) {
             throw new IllegalArgumentException("entity must not be null");
         }
@@ -123,7 +123,7 @@ public static String loadAndRun(String base64EncodedEntityRequest, ITaskEntity e
      * @throws IllegalArgumentException if either parameter is {@code null} or if {@code entityRequestBytes}
      *                                  is not valid protobuf
      */
-    public static byte[] loadAndRun(byte[] entityRequestBytes, ITaskEntity entity) {
+    public static byte[] loadAndRun(byte[] entityRequestBytes, TaskEntity entity) {
         if (entity == null) {
             throw new IllegalArgumentException("entity must not be null");
         }