#188

Committed 25 Sep 2023 04:42PM UTC coverage: 77.369% (-0.3%) from 77.663%

Build # #188

Build Type

push

github-actions

Committed by

web-flow

Commit Message

Fix null pointer on trigger immediately (#1865)

Run Details

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

18670 of 24131 relevant lines covered (77.37%)

0.77 hits per line

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

77.27

/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,
      io.temporal.api.common.v1.Header header) {
    try {
      context.handleSignal(signalName, input, eventId, new Header(header));
    } catch (Throwable e) {
      applyWorkflowFailurePolicyAndRethrow(e);
    }
  }

  public Optional<Payloads> handleQuery(
      String type, io.temporal.api.common.v1.Header header, Optional<Payloads> args) {
    return context.handleQuery(type, new Header(header), args);
  }

  public void handleValidateUpdate(
      String updateName,
      Optional<Payloads> input,
      long eventId,
      io.temporal.api.common.v1.Header header) {
    try {
      context.handleValidateUpdate(updateName, input, eventId, new Header(header));
    } catch (Throwable e) {
      applyWorkflowFailurePolicyAndRethrow(e);
    }
  }

  public Optional<Payloads> handleExecuteUpdate(
      String updateName,
      Optional<Payloads> input,
      long eventId,
      io.temporal.api.common.v1.Header header) {
    try {
      return context.handleExecuteUpdate(updateName, input, eventId, new Header(header));
    } 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(
91	String signalName,
92	Optional<Payloads> input,
93	long eventId,
94	io.temporal.api.common.v1.Header header) {
95	try {
96	context.handleSignal(signalName, input, eventId, new Header(header));	1✔
97	} catch (Throwable e) {	1✔
98	applyWorkflowFailurePolicyAndRethrow(e);	×
99	}	1✔
100	}	1✔
101
102	public Optional<Payloads> handleQuery(
103	String type, io.temporal.api.common.v1.Header header, Optional<Payloads> args) {
104	return context.handleQuery(type, new Header(header), args);	1✔
105	}
106
107	public void handleValidateUpdate(
108	String updateName,
109	Optional<Payloads> input,
110	long eventId,
111	io.temporal.api.common.v1.Header header) {
112	try {
113	context.handleValidateUpdate(updateName, input, eventId, new Header(header));	1✔
114	} catch (Throwable e) {	1✔
115	applyWorkflowFailurePolicyAndRethrow(e);	×
116	}	1✔
117	}	1✔
118
119	public Optional<Payloads> handleExecuteUpdate(
120	String updateName,
121	Optional<Payloads> input,
122	long eventId,
123	io.temporal.api.common.v1.Header header) {
124	try {
125	return context.handleExecuteUpdate(updateName, input, eventId, new Header(header));	1✔
126	} catch (Throwable e) {	1✔
127	applyWorkflowFailurePolicyAndRethrow(e);	×
128	}
129	return Optional.empty();	×
130	}
131
132	private void applyWorkflowFailurePolicyAndRethrow(Throwable e) {
133	if (e instanceof DestroyWorkflowThreadError) {	1✔
134	throw (DestroyWorkflowThreadError) e;	1✔
135	}
136	Throwable exception = unwrap(e);	1✔
137
138	Class<? extends Throwable>[] failTypes = implementationOptions.getFailWorkflowExceptionTypes();	1✔
139	if (exception instanceof TemporalFailure) {	1✔
140	throwAndFailWorkflowExecution(exception);	×
141	}
142	for (Class<? extends Throwable> failType : failTypes) {	1✔
143	if (failType.isAssignableFrom(exception.getClass())) {	1✔
144	throwAndFailWorkflowExecution(exception);	×
145	}
146	}
147
148	throw wrap(exception);	1✔
149	}
150
151	private void throwAndFailWorkflowExecution(Throwable exception) {
152	ReplayWorkflowContext replayWorkflowContext = context.getReplayContext();	1✔
153	@Nullable
154	String fullReplayDirectQueryName = replayWorkflowContext.getFullReplayDirectQueryName();	1✔
155	WorkflowInfo info = Workflow.getInfo();	1✔
156
157	if (fullReplayDirectQueryName != null) {	1✔
158	if (log.isDebugEnabled()	1✔
159	&& !requestedCancellation(replayWorkflowContext.isCancelRequested(), exception)) {	×
160	log.debug(	×
161	"Replayed workflow execution failure WorkflowId='{}', RunId={}, WorkflowType='{}' for direct query QueryType='{}'",
162	info.getWorkflowId(),	×
163	info.getRunId(),	×
164	info.getWorkflowType(),	×
165	fullReplayDirectQueryName,
166	exception);
167	}
168	} else {
169	if (log.isWarnEnabled()	1✔
170	&& !requestedCancellation(replayWorkflowContext.isCancelRequested(), exception)) {	1✔
171	log.warn(	1✔
172	"Workflow execution failure WorkflowId='{}', RunId={}, WorkflowType='{}'",
173	info.getWorkflowId(),	1✔
174	info.getRunId(),	1✔
175	info.getWorkflowType(),	1✔
176	exception);
177	}
178	}
179
180	throw new WorkflowExecutionException(context.mapWorkflowExceptionToFailure(exception));	1✔
181	}
182
183	/**
184	* @return true if both workflow cancellation is requested and the exception contains a
185	* cancellation exception in the chain
186	*/
187	private boolean requestedCancellation(boolean cancelRequested, Throwable exception) {
188	return cancelRequested && isCanceledCause(exception);	1✔
189	}
190
191	private static boolean isCanceledCause(Throwable exception) {
192	while (exception != null) {	1✔
193	if (exception instanceof CanceledFailure) {	1✔
194	return true;	1✔
195	}
196	exception = exception.getCause();	1✔
197	}
198	return false;	×
199	}
200	}

temporalio / sdk-java / #188

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