#181

pending completion

Build # #181

Build Type

push

github-actions

Committed by

web-flow

Commit Message

Properly wrap exceptions from schedule client (#1827)

Wrap schedule exception

Run Details

37 of 37 new or added lines in 1 file covered. (100.0%)

18557 of 23894 relevant lines covered (77.66%)

0.78 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.71

/temporal-sdk/src/main/java/io/temporal/internal/sync/ChildWorkflowStubImpl.java

/*
 * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
 *
 * Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
 *
 * Modifications copyright (C) 2017 Uber Technologies, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this material except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.temporal.internal.sync;

import com.google.common.base.Defaults;
import io.temporal.api.common.v1.WorkflowExecution;
import io.temporal.common.interceptors.Header;
import io.temporal.common.interceptors.WorkflowOutboundCallsInterceptor;
import io.temporal.common.interceptors.WorkflowOutboundCallsInterceptor.ChildWorkflowOutput;
import io.temporal.failure.TemporalFailure;
import io.temporal.workflow.*;
import java.lang.reflect.Type;
import java.util.Objects;

class ChildWorkflowStubImpl implements ChildWorkflowStub {

  private final String workflowType;
  private final ChildWorkflowOptions options;
  private final WorkflowOutboundCallsInterceptor outboundCallsInterceptor;
  private final CompletablePromise<WorkflowExecution> execution;
  private final Functions.Proc1<String> assertReadOnly;

  ChildWorkflowStubImpl(
      String workflowType,
      ChildWorkflowOptions options,
      WorkflowOutboundCallsInterceptor outboundCallsInterceptor,
      Functions.Proc1<String> assertReadOnly) {
    this.workflowType = Objects.requireNonNull(workflowType);
    this.options = ChildWorkflowOptions.newBuilder(options).validateAndBuildWithDefaults();
    this.outboundCallsInterceptor = Objects.requireNonNull(outboundCallsInterceptor);
    this.execution = Workflow.newPromise();
    // We register an empty handler to make sure that this promise is always "accessed" and never
    // leads to a log about it being completed exceptionally and non-accessed.
    // The "main" Child Workflow promise is the one returned from the execute method and that
    // promise will always be logged if not accessed.
    this.execution.handle((ex, failure) -> null);
    this.assertReadOnly = assertReadOnly;
  }

  @Override
  public String getWorkflowType() {
    return workflowType;
  }

  @Override
  public Promise<WorkflowExecution> getExecution() {
    // We create a new Promise here because we want it to be registered with the Runner
    CompletablePromise<WorkflowExecution> result = Workflow.newPromise();
    result.completeFrom(this.execution);
    return result;
  }

  @Override
  public ChildWorkflowOptions getOptions() {
    return options;
  }

  @Override
  public <R> R execute(Class<R> resultClass, Object... args) {
    return execute(resultClass, resultClass, args);
  }

  @Override
  public <R> R execute(Class<R> resultClass, Type resultType, Object... args) {
    assertReadOnly.apply("schedule child workflow");
    Promise<R> result = executeAsync(resultClass, resultType, args);
    if (AsyncInternal.isAsync()) {
      AsyncInternal.setAsyncResult(result);
      return Defaults.defaultValue(resultClass);
    }
    try {
      return result.get();
    } catch (TemporalFailure e) {
      // Reset stack to the current one. Otherwise it is very confusing to see a stack of
      // an event handling method.
      e.setStackTrace(Thread.currentThread().getStackTrace());
      throw e;
    }
  }

  @Override
  public <R> Promise<R> executeAsync(Class<R> resultClass, Object... args) {
    return executeAsync(resultClass, resultClass, args);
  }

  @Override
  public <R> Promise<R> executeAsync(Class<R> resultClass, Type resultType, Object... args) {
    assertReadOnly.apply("schedule child workflow");
    ChildWorkflowOutput<R> result =
        outboundCallsInterceptor.executeChildWorkflow(
            new WorkflowOutboundCallsInterceptor.ChildWorkflowInput<>(
                getWorkflowIdForStart(options),
                workflowType,
                resultClass,
                resultType,
                args,
                options,
                Header.empty()));
    execution.completeFrom(result.getWorkflowExecution());
    return result.getResult();
  }

  @Override
  public void signal(String signalName, Object... args) {
    assertReadOnly.apply("signal workflow");
    Promise<Void> signaled =
        outboundCallsInterceptor
            .signalExternalWorkflow(
                new WorkflowOutboundCallsInterceptor.SignalExternalInput(
                    execution.get(), signalName, args))
            .getResult();
    if (AsyncInternal.isAsync()) {
      AsyncInternal.setAsyncResult(signaled);
      return;
    }
    try {
      signaled.get();
    } catch (SignalExternalWorkflowException e) {
      // Reset stack to the current one. Otherwise it is very confusing to see a stack of
      // an event handling method.
      e.setStackTrace(Thread.currentThread().getStackTrace());
      throw e;
    }
  }

  private String getWorkflowIdForStart(ChildWorkflowOptions options) {
    String workflowId = options.getWorkflowId();
    if (workflowId == null) {
      workflowId = Workflow.randomUUID().toString();
    }
    return workflowId;
  }
}

1	/*
2	* Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
3	*
4	* Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
5	*
6	* Modifications copyright (C) 2017 Uber Technologies, Inc.
7	*
8	* Licensed under the Apache License, Version 2.0 (the "License");
9	* you may not use this material except in compliance with the License.
10	* You may obtain a copy of the License at
11	*
12	* http://www.apache.org/licenses/LICENSE-2.0
13	*
14	* Unless required by applicable law or agreed to in writing, software
15	* distributed under the License is distributed on an "AS IS" BASIS,
16	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17	* See the License for the specific language governing permissions and
18	* limitations under the License.
19	*/
20
21	package io.temporal.internal.sync;
22
23	import com.google.common.base.Defaults;
24	import io.temporal.api.common.v1.WorkflowExecution;
25	import io.temporal.common.interceptors.Header;
26	import io.temporal.common.interceptors.WorkflowOutboundCallsInterceptor;
27	import io.temporal.common.interceptors.WorkflowOutboundCallsInterceptor.ChildWorkflowOutput;
28	import io.temporal.failure.TemporalFailure;
29	import io.temporal.workflow.*;
30	import java.lang.reflect.Type;
31	import java.util.Objects;
32
33	class ChildWorkflowStubImpl implements ChildWorkflowStub {
34
35	private final String workflowType;
36	private final ChildWorkflowOptions options;
37	private final WorkflowOutboundCallsInterceptor outboundCallsInterceptor;
38	private final CompletablePromise<WorkflowExecution> execution;
39	private final Functions.Proc1<String> assertReadOnly;
40
41	ChildWorkflowStubImpl(
42	String workflowType,
43	ChildWorkflowOptions options,
44	WorkflowOutboundCallsInterceptor outboundCallsInterceptor,
45	Functions.Proc1<String> assertReadOnly) {	1✔
46	this.workflowType = Objects.requireNonNull(workflowType);	1✔
47	this.options = ChildWorkflowOptions.newBuilder(options).validateAndBuildWithDefaults();	1✔
48	this.outboundCallsInterceptor = Objects.requireNonNull(outboundCallsInterceptor);	1✔
49	this.execution = Workflow.newPromise();	1✔
50	// We register an empty handler to make sure that this promise is always "accessed" and never
51	// leads to a log about it being completed exceptionally and non-accessed.
52	// The "main" Child Workflow promise is the one returned from the execute method and that
53	// promise will always be logged if not accessed.
54	this.execution.handle((ex, failure) -> null);	1✔
55	this.assertReadOnly = assertReadOnly;	1✔
56	}	1✔
57
58	@Override
59	public String getWorkflowType() {
60	return workflowType;	×
61	}
62
63	@Override
64	public Promise<WorkflowExecution> getExecution() {
65	// We create a new Promise here because we want it to be registered with the Runner
66	CompletablePromise<WorkflowExecution> result = Workflow.newPromise();	1✔
67	result.completeFrom(this.execution);	1✔
68	return result;	1✔
69	}
70
71	@Override
72	public ChildWorkflowOptions getOptions() {
73	return options;	×
74	}
75
76	@Override
77	public <R> R execute(Class<R> resultClass, Object... args) {
78	return execute(resultClass, resultClass, args);	1✔
79	}
80
81	@Override
82	public <R> R execute(Class<R> resultClass, Type resultType, Object... args) {
83	assertReadOnly.apply("schedule child workflow");	1✔
84	Promise<R> result = executeAsync(resultClass, resultType, args);	1✔
85	if (AsyncInternal.isAsync()) {	1✔
86	AsyncInternal.setAsyncResult(result);	1✔
87	return Defaults.defaultValue(resultClass);	1✔
88	}
89	try {
90	return result.get();	1✔
91	} catch (TemporalFailure e) {	1✔
92	// Reset stack to the current one. Otherwise it is very confusing to see a stack of
93	// an event handling method.
94	e.setStackTrace(Thread.currentThread().getStackTrace());	1✔
95	throw e;	1✔
96	}
97	}
98
99	@Override
100	public <R> Promise<R> executeAsync(Class<R> resultClass, Object... args) {
101	return executeAsync(resultClass, resultClass, args);	1✔
102	}
103
104	@Override
105	public <R> Promise<R> executeAsync(Class<R> resultClass, Type resultType, Object... args) {
106	assertReadOnly.apply("schedule child workflow");	1✔
107	ChildWorkflowOutput<R> result =	1✔
108	outboundCallsInterceptor.executeChildWorkflow(	1✔
109	new WorkflowOutboundCallsInterceptor.ChildWorkflowInput<>(
110	getWorkflowIdForStart(options),	1✔
111	workflowType,
112	resultClass,
113	resultType,
114	args,
115	options,
116	Header.empty()));	1✔
117	execution.completeFrom(result.getWorkflowExecution());	1✔
118	return result.getResult();	1✔
119	}
120
121	@Override
122	public void signal(String signalName, Object... args) {
123	assertReadOnly.apply("signal workflow");	1✔
124	Promise<Void> signaled =	1✔
125	outboundCallsInterceptor
126	.signalExternalWorkflow(	1✔
127	new WorkflowOutboundCallsInterceptor.SignalExternalInput(
128	execution.get(), signalName, args))	1✔
129	.getResult();	1✔
130	if (AsyncInternal.isAsync()) {	1✔
131	AsyncInternal.setAsyncResult(signaled);	×
132	return;	×
133	}
134	try {
135	signaled.get();	1✔
136	} catch (SignalExternalWorkflowException e) {	×
137	// Reset stack to the current one. Otherwise it is very confusing to see a stack of
138	// an event handling method.
139	e.setStackTrace(Thread.currentThread().getStackTrace());	×
140	throw e;	×
141	}	1✔
142	}	1✔
143
144	private String getWorkflowIdForStart(ChildWorkflowOptions options) {
145	String workflowId = options.getWorkflowId();	1✔
146	if (workflowId == null) {	1✔
147	workflowId = Workflow.randomUUID().toString();	1✔
148	}
149	return workflowId;	1✔
150	}
151	}

temporalio / sdk-java / #181

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous