fix: create_orchestration span kind to PRODUCER, error status on failed orchestration

1. Changed create_orchestration span from INTERNAL (default) to PRODUCER,
   matching .NET SDK's StartActivityForNewOrchestration which uses
   ActivityKind.Producer.

2. Set ERROR status on orchestration span when the orchestration fails,
   matching .NET SDK's pattern of checking CompleteOrchestration action
   for FAILED status before disposing the span.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: torosent

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

@@ -10,6 +10,7 @@
 
 import io.grpc.*;
 import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanKind;
 import io.opentelemetry.context.Scope;
 
 import javax.annotation.Nullable;
@@ -144,7 +145,7 @@ public String scheduleNewOrchestrationInstance(
         spanAttrs.put(TracingHelper.ATTR_INSTANCE_ID, instanceId);
         Span createSpan = TracingHelper.startSpan(
                 TracingHelper.TYPE_CREATE_ORCHESTRATION + ":" + orchestratorName,
-                null, null, spanAttrs);
+                null, SpanKind.PRODUCER, spanAttrs);
         Scope createScope = createSpan.makeCurrent();
 
         try {

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

@@ -13,6 +13,7 @@
 import io.grpc.*;
 import io.opentelemetry.api.trace.Span;
 import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.api.trace.StatusCode;
 import io.opentelemetry.context.Scope;
 
 import java.time.Duration;
@@ -231,8 +232,20 @@ public void startAndBlock() {
                                 throw new RuntimeException(e);
                             }
 
-                            // End the orchestration span for every dispatch
+                            // End the orchestration span, setting error status if the orchestration failed
                             if (orchestrationSpan != null) {
+                                for (OrchestratorAction action : taskOrchestratorResult.getActions()) {
+                                    if (action.getOrchestratorActionTypeCase() == OrchestratorAction.OrchestratorActionTypeCase.COMPLETEORCHESTRATION) {
+                                        CompleteOrchestrationAction complete = action.getCompleteOrchestration();
+                                        if (complete.getOrchestrationStatus() == OrchestrationStatus.ORCHESTRATION_STATUS_FAILED) {
+                                            String errorMsg = complete.hasFailureDetails()
+                                                    ? complete.getFailureDetails().getErrorMessage()
+                                                    : "Orchestration failed";
+                                            orchestrationSpan.setStatus(StatusCode.ERROR, errorMsg);
+                                        }
+                                        break;
+                                    }
+                                }
                                 orchestrationSpan.end();
                             }