#169

pending completion

Build # #169

Build Type

push

github-actions

Committed by

web-flow

Commit Message

Remove use of deprecated API (#1758)

Run Details

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

17345 of 21558 relevant lines covered (80.46%)

0.8 hits per line

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

68.18

/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowExecutionHandler.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 static io.temporal.internal.sync.WorkflowInternal.unwrap;
import static io.temporal.serviceclient.CheckedExceptionWrapper.wrap;

import io.temporal.api.common.v1.Payloads;
import io.temporal.api.history.v1.WorkflowExecutionStartedEventAttributes;
import io.temporal.common.interceptors.Header;
import io.temporal.failure.CanceledFailure;
import io.temporal.failure.TemporalFailure;
import io.temporal.internal.replay.ReplayWorkflowContext;
import io.temporal.internal.worker.WorkflowExecutionException;
import io.temporal.worker.WorkflowImplementationOptions;
import io.temporal.workflow.Workflow;
import io.temporal.workflow.WorkflowInfo;
import java.util.Objects;
import java.util.Optional;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

class WorkflowExecutionHandler {

  private static final Logger log = LoggerFactory.getLogger(WorkflowExecutionHandler.class);

  private final SyncWorkflowContext context;
  private final SyncWorkflowDefinition workflow;
  private final WorkflowExecutionStartedEventAttributes attributes;
  @Nonnull private final WorkflowImplementationOptions implementationOptions;

  private Optional<Payloads> output = Optional.empty();
  private boolean done;

  public WorkflowExecutionHandler(
      SyncWorkflowContext context,
      SyncWorkflowDefinition workflow,
      WorkflowExecutionStartedEventAttributes attributes,
      @Nonnull WorkflowImplementationOptions options) {
    this.implementationOptions = options;
    this.context = Objects.requireNonNull(context);
    this.workflow = Objects.requireNonNull(workflow);
    this.attributes = Objects.requireNonNull(attributes);
  }

  public void runWorkflowMethod() {
    try {
      Optional<Payloads> input =
          attributes.hasInput() ? Optional.of(attributes.getInput()) : Optional.empty();
      output = workflow.execute(new Header(attributes.getHeader()), input);
    } catch (Throwable e) {
      applyWorkflowFailurePolicyAndRethrow(e);
    } finally {
      done = true;
    }
  }

  public void cancel(String reason) {}

  public boolean isDone() {
    return done;
  }

  public Optional<Payloads> getOutput() {
    return output;
  }

  public void close() {}

  public void handleSignal(String signalName, Optional<Payloads> input, long eventId) {
    try {
      context.handleSignal(signalName, input, eventId);
    } catch (Throwable e) {
      applyWorkflowFailurePolicyAndRethrow(e);
    }
  }

  public Optional<Payloads> handleQuery(String type, Optional<Payloads> args) {
    return context.handleQuery(type, args);
  }

  public void handleValidateUpdate(String updateName, Optional<Payloads> input, long eventId) {
    try {
      context.handleValidateUpdate(updateName, input, eventId);
    } catch (Throwable e) {
      applyWorkflowFailurePolicyAndRethrow(e);
    }
  }

  public Optional<Payloads> handleExecuteUpdate(
      String updateName, Optional<Payloads> input, long eventId) {
    try {
      return context.handleExecuteUpdate(updateName, input, eventId);
    } catch (Throwable e) {
      applyWorkflowFailurePolicyAndRethrow(e);
    }
    return Optional.empty();
  }

  private void applyWorkflowFailurePolicyAndRethrow(Throwable e) {
    if (e instanceof DestroyWorkflowThreadError) {
      throw (DestroyWorkflowThreadError) e;
    }
    Throwable exception = unwrap(e);

    Class<? extends Throwable>[] failTypes = implementationOptions.getFailWorkflowExceptionTypes();
    if (exception instanceof TemporalFailure) {
      throwAndFailWorkflowExecution(exception);
    }
    for (Class<? extends Throwable> failType : failTypes) {
      if (failType.isAssignableFrom(exception.getClass())) {
        throwAndFailWorkflowExecution(exception);
      }
    }

    throw wrap(exception);
  }

  private void throwAndFailWorkflowExecution(Throwable exception) {
    ReplayWorkflowContext replayWorkflowContext = context.getReplayContext();
    @Nullable
    String fullReplayDirectQueryName = replayWorkflowContext.getFullReplayDirectQueryName();
    WorkflowInfo info = Workflow.getInfo();

    if (fullReplayDirectQueryName != null) {
      if (log.isDebugEnabled()
          && !requestedCancellation(replayWorkflowContext.isCancelRequested(), exception)) {
        log.debug(
            "Replayed workflow execution failure WorkflowId='{}', RunId={}, WorkflowType='{}' for direct query QueryType='{}'",
            info.getWorkflowId(),
            info.getRunId(),
            info.getWorkflowType(),
            fullReplayDirectQueryName,
            exception);
      }
    } else {
      if (log.isWarnEnabled()
          && !requestedCancellation(replayWorkflowContext.isCancelRequested(), exception)) {
        log.warn(
            "Workflow execution failure WorkflowId='{}', RunId={}, WorkflowType='{}'",
            info.getWorkflowId(),
            info.getRunId(),
            info.getWorkflowType(),
            exception);
      }
    }

    throw new WorkflowExecutionException(context.mapWorkflowExceptionToFailure(exception));
  }

  /**
   * @return true if both workflow cancellation is requested and the exception contains a
   *     cancellation exception in the chain
   */
  private boolean requestedCancellation(boolean cancelRequested, Throwable exception) {
    return cancelRequested && isCanceledCause(exception);
  }

  private static boolean isCanceledCause(Throwable exception) {
    while (exception != null) {
      if (exception instanceof CanceledFailure) {
        return true;
      }
      exception = exception.getCause();
    }
    return false;
  }
}

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 static io.temporal.internal.sync.WorkflowInternal.unwrap;
24	import static io.temporal.serviceclient.CheckedExceptionWrapper.wrap;
25
26	import io.temporal.api.common.v1.Payloads;
27	import io.temporal.api.history.v1.WorkflowExecutionStartedEventAttributes;
28	import io.temporal.common.interceptors.Header;
29	import io.temporal.failure.CanceledFailure;
30	import io.temporal.failure.TemporalFailure;
31	import io.temporal.internal.replay.ReplayWorkflowContext;
32	import io.temporal.internal.worker.WorkflowExecutionException;
33	import io.temporal.worker.WorkflowImplementationOptions;
34	import io.temporal.workflow.Workflow;
35	import io.temporal.workflow.WorkflowInfo;
36	import java.util.Objects;
37	import java.util.Optional;
38	import javax.annotation.Nonnull;
39	import javax.annotation.Nullable;
40	import org.slf4j.Logger;
41	import org.slf4j.LoggerFactory;
42
43	class WorkflowExecutionHandler {
44
45	private static final Logger log = LoggerFactory.getLogger(WorkflowExecutionHandler.class);	1✔
46
47	private final SyncWorkflowContext context;
48	private final SyncWorkflowDefinition workflow;
49	private final WorkflowExecutionStartedEventAttributes attributes;
50	@Nonnull private final WorkflowImplementationOptions implementationOptions;
51
52	private Optional<Payloads> output = Optional.empty();	1✔
53	private boolean done;
54
55	public WorkflowExecutionHandler(
56	SyncWorkflowContext context,
57	SyncWorkflowDefinition workflow,
58	WorkflowExecutionStartedEventAttributes attributes,
59	@Nonnull WorkflowImplementationOptions options) {	1✔
60	this.implementationOptions = options;	1✔
61	this.context = Objects.requireNonNull(context);	1✔
62	this.workflow = Objects.requireNonNull(workflow);	1✔
63	this.attributes = Objects.requireNonNull(attributes);	1✔
64	}	1✔
65
66	public void runWorkflowMethod() {
67	try {
68	Optional<Payloads> input =
69	attributes.hasInput() ? Optional.of(attributes.getInput()) : Optional.empty();	1✔
70	output = workflow.execute(new Header(attributes.getHeader()), input);	1✔
71	} catch (Throwable e) {	1✔
72	applyWorkflowFailurePolicyAndRethrow(e);	×
73	} finally {
74	done = true;	1✔
75	}
76	}	1✔
77
78	public void cancel(String reason) {}	×
79
80	public boolean isDone() {
81	return done;	1✔
82	}
83
84	public Optional<Payloads> getOutput() {
85	return output;	1✔
86	}
87
88	public void close() {}	×
89
90	public void handleSignal(String signalName, Optional<Payloads> input, long eventId) {
91	try {
92	context.handleSignal(signalName, input, eventId);	1✔
93	} catch (Throwable e) {	1✔
94	applyWorkflowFailurePolicyAndRethrow(e);	×
95	}	1✔
96	}	1✔
97
98	public Optional<Payloads> handleQuery(String type, Optional<Payloads> args) {
99	return context.handleQuery(type, args);	1✔
100	}
101
102	public void handleValidateUpdate(String updateName, Optional<Payloads> input, long eventId) {
103	try {
104	context.handleValidateUpdate(updateName, input, eventId);	×
105	} catch (Throwable e) {	×
106	applyWorkflowFailurePolicyAndRethrow(e);	×
107	}	×
108	}	×
109
110	public Optional<Payloads> handleExecuteUpdate(
111	String updateName, Optional<Payloads> input, long eventId) {
112	try {
113	return context.handleExecuteUpdate(updateName, input, eventId);	×
114	} catch (Throwable e) {	×
115	applyWorkflowFailurePolicyAndRethrow(e);	×
116	}
117	return Optional.empty();	×
118	}
119
120	private void applyWorkflowFailurePolicyAndRethrow(Throwable e) {
121	if (e instanceof DestroyWorkflowThreadError) {	1✔
122	throw (DestroyWorkflowThreadError) e;	1✔
123	}
124	Throwable exception = unwrap(e);	1✔
125
126	Class<? extends Throwable>[] failTypes = implementationOptions.getFailWorkflowExceptionTypes();	1✔
127	if (exception instanceof TemporalFailure) {	1✔
128	throwAndFailWorkflowExecution(exception);	×
129	}
130	for (Class<? extends Throwable> failType : failTypes) {	1✔
131	if (failType.isAssignableFrom(exception.getClass())) {	1✔
132	throwAndFailWorkflowExecution(exception);	×
133	}
134	}
135
136	throw wrap(exception);	1✔
137	}
138
139	private void throwAndFailWorkflowExecution(Throwable exception) {
140	ReplayWorkflowContext replayWorkflowContext = context.getReplayContext();	1✔
141	@Nullable
142	String fullReplayDirectQueryName = replayWorkflowContext.getFullReplayDirectQueryName();	1✔
143	WorkflowInfo info = Workflow.getInfo();	1✔
144
145	if (fullReplayDirectQueryName != null) {	1✔
146	if (log.isDebugEnabled()	1✔
147	&& !requestedCancellation(replayWorkflowContext.isCancelRequested(), exception)) {	×
148	log.debug(	×
149	"Replayed workflow execution failure WorkflowId='{}', RunId={}, WorkflowType='{}' for direct query QueryType='{}'",
150	info.getWorkflowId(),	×
151	info.getRunId(),	×
152	info.getWorkflowType(),	×
153	fullReplayDirectQueryName,
154	exception);
155	}
156	} else {
157	if (log.isWarnEnabled()	1✔
158	&& !requestedCancellation(replayWorkflowContext.isCancelRequested(), exception)) {	1✔
159	log.warn(	1✔
160	"Workflow execution failure WorkflowId='{}', RunId={}, WorkflowType='{}'",
161	info.getWorkflowId(),	1✔
162	info.getRunId(),	1✔
163	info.getWorkflowType(),	1✔
164	exception);
165	}
166	}
167
168	throw new WorkflowExecutionException(context.mapWorkflowExceptionToFailure(exception));	1✔
169	}
170
171	/**
172	* @return true if both workflow cancellation is requested and the exception contains a
173	* cancellation exception in the chain
174	*/
175	private boolean requestedCancellation(boolean cancelRequested, Throwable exception) {
176	return cancelRequested && isCanceledCause(exception);	1✔
177	}
178
179	private static boolean isCanceledCause(Throwable exception) {
180	while (exception != null) {	1✔
181	if (exception instanceof CanceledFailure) {	1✔
182	return true;	1✔
183	}
184	exception = exception.getCause();	1✔
185	}
186	return false;	×
187	}
188	}

temporalio / sdk-java / #169

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