#103

pending completion

Build # #103

Build Type

push

github-actions

Committed by

web-flow

Commit Message

Implement retry of local activities for over local retry threshold duration (#1542)

Issue #1261

Run Details

244 of 244 new or added lines in 16 files covered. (100.0%)

16122 of 19841 relevant lines covered (81.26%)

0.81 hits per line

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

97.3

/temporal-sdk/src/main/java/io/temporal/internal/worker/LocalActivityExecutionContext.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.worker;

import io.grpc.Deadline;
import io.temporal.api.failure.v1.Failure;
import io.temporal.api.workflow.v1.PendingActivityInfo;
import io.temporal.api.workflowservice.v1.DescribeWorkflowExecutionRequest;
import io.temporal.api.workflowservice.v1.PollActivityTaskQueueResponse;
import io.temporal.internal.statemachines.ExecuteLocalActivityParameters;
import io.temporal.workflow.Functions;
import java.time.Duration;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;

class LocalActivityExecutionContext {
  private final @Nonnull ExecuteLocalActivityParameters executionParams;
  private final @Nonnull CompletableFuture<LocalActivityResult> executionResult =
      new CompletableFuture<>();

  private final @Nullable Deadline scheduleToCloseDeadline;

  private final @Nonnull AtomicInteger currentAttempt;

  private final @Nonnull AtomicReference<Failure> lastAttemptFailure = new AtomicReference<>();

  private @Nullable ScheduledFuture<?> scheduleToCloseFuture;

  public LocalActivityExecutionContext(
      @Nonnull ExecuteLocalActivityParameters executionParams,
      @Nonnull Functions.Proc1<LocalActivityResult> resultCallback,
      @Nullable Deadline scheduleToCloseDeadline) {
    this.executionParams = Objects.requireNonNull(executionParams, "executionParams");
    this.executionResult.thenAccept(
        Objects.requireNonNull(resultCallback, "resultCallback")::apply);
    this.scheduleToCloseDeadline = scheduleToCloseDeadline;
    this.currentAttempt = new AtomicInteger(executionParams.getInitialAttempt());
    Failure previousExecutionFailure = executionParams.getPreviousLocalExecutionFailure();
    if (previousExecutionFailure != null) {
      if (previousExecutionFailure.hasTimeoutFailureInfo() && previousExecutionFailure.hasCause()) {
        // It can be only startToClose timeout (others would be fatal).
        // Structure TIMEOUT_TYPE_START_TO_CLOSE -> TIMEOUT_TYPE_START_TO_CLOSE or
        // TIMEOUT_TYPE_START_TO_CLOSE -> ApplicationFailure is possible here - chaining of
        // startToClose timeout with a previous failure.
        // We reconstruct the last attempt failure (that would typically be preserved in a mutable
        // state) from local activity execution failure.
        // See a similar logic in
        // io.temporal.internal.testservice.StateMachines#timeoutActivityTask.
        lastAttemptFailure.set(Failure.newBuilder(previousExecutionFailure).clearCause().build());
      } else {
        lastAttemptFailure.set(previousExecutionFailure);
      }
    }
  }

  public String getActivityId() {
    return executionParams.getActivityId();
  }

  public int getCurrentAttempt() {
    return currentAttempt.get();
  }

  /**
   * The last failure preserved for this activity execution. This field is mimicking the behavior of
   * {@link PendingActivityInfo#getLastFailure()} that is maintained by the server in the mutable
   * state and is used to create ActivityFailures with meaningful causes and returned by {@link
   * io.temporal.api.workflowservice.v1.WorkflowServiceGrpc.WorkflowServiceBlockingStub#describeWorkflowExecution(DescribeWorkflowExecutionRequest)}
   */
  @Nullable
  public Failure getLastAttemptFailure() {
    return lastAttemptFailure.get();
  }

  @Nullable
  public Failure getPreviousExecutionFailure() {
    return executionParams.getPreviousLocalExecutionFailure();
  }

  @Nonnull
  public PollActivityTaskQueueResponse.Builder getInitialTask() {
    return executionParams.cloneActivityTaskBuilder();
  }

  @Nonnull
  public PollActivityTaskQueueResponse.Builder getNextAttemptActivityTask(
      @Nullable Failure lastFailure) {
    // synchronization here is not absolutely needed as LocalActivityWorker#scheduleNextAttempt
    // shouldn't be executed concurrently. But to make sure this code is safe for future changes,
    // let's make this method atomic and protect thread-unsafe protobuf builder modification.

    // executionResult here is used just as an internal monitor object that is final and never
    // escapes the class
    int nextAttempt;
    synchronized (executionResult) {
      nextAttempt = currentAttempt.incrementAndGet();
      if (lastFailure != null) {
        this.lastAttemptFailure.set(lastFailure);
      }
    }
    // doesn't need to be synchronized as we clone instead of modifying the original task builder
    return executionParams
        .cloneActivityTaskBuilder()
        .setAttempt(nextAttempt)
        .clearCurrentAttemptScheduledTime();
  }

  @Nullable
  public Deadline getScheduleToCloseDeadline() {
    return scheduleToCloseDeadline;
  }

  public void setScheduleToCloseFuture(@Nullable ScheduledFuture<?> scheduleToCloseFuture) {
    this.scheduleToCloseFuture = scheduleToCloseFuture;
  }

  public boolean callback(LocalActivityResult result) {
    if (scheduleToCloseFuture != null) {
      scheduleToCloseFuture.cancel(false);
    }
    return executionResult.complete(result);
  }

  @Nullable
  public Duration getScheduleToStartTimeout() {
    return executionParams.getScheduleToStartTimeout();
  }

  public long getOriginalScheduledTimestamp() {
    return executionParams.getOriginalScheduledTimestamp();
  }

  @Nonnull
  public Duration getLocalRetryThreshold() {
    return executionParams.getLocalRetryThreshold();
  }
}

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.worker;
22
23	import io.grpc.Deadline;
24	import io.temporal.api.failure.v1.Failure;
25	import io.temporal.api.workflow.v1.PendingActivityInfo;
26	import io.temporal.api.workflowservice.v1.DescribeWorkflowExecutionRequest;
27	import io.temporal.api.workflowservice.v1.PollActivityTaskQueueResponse;
28	import io.temporal.internal.statemachines.ExecuteLocalActivityParameters;
29	import io.temporal.workflow.Functions;
30	import java.time.Duration;
31	import java.util.Objects;
32	import java.util.concurrent.CompletableFuture;
33	import java.util.concurrent.ScheduledFuture;
34	import java.util.concurrent.atomic.AtomicInteger;
35	import java.util.concurrent.atomic.AtomicReference;
36	import javax.annotation.Nonnull;
37	import javax.annotation.Nullable;
38
39	class LocalActivityExecutionContext {
40	private final @Nonnull ExecuteLocalActivityParameters executionParams;
41	private final @Nonnull CompletableFuture<LocalActivityResult> executionResult =	1✔
42	new CompletableFuture<>();
43
44	private final @Nullable Deadline scheduleToCloseDeadline;
45
46	private final @Nonnull AtomicInteger currentAttempt;
47
48	private final @Nonnull AtomicReference<Failure> lastAttemptFailure = new AtomicReference<>();	1✔
49
50	private @Nullable ScheduledFuture<?> scheduleToCloseFuture;
51
52	public LocalActivityExecutionContext(
53	@Nonnull ExecuteLocalActivityParameters executionParams,
54	@Nonnull Functions.Proc1<LocalActivityResult> resultCallback,
55	@Nullable Deadline scheduleToCloseDeadline) {	1✔
56	this.executionParams = Objects.requireNonNull(executionParams, "executionParams");	1✔
57	this.executionResult.thenAccept(	1✔
58	Objects.requireNonNull(resultCallback, "resultCallback")::apply);	1✔
59	this.scheduleToCloseDeadline = scheduleToCloseDeadline;	1✔
60	this.currentAttempt = new AtomicInteger(executionParams.getInitialAttempt());	1✔
61	Failure previousExecutionFailure = executionParams.getPreviousLocalExecutionFailure();	1✔
62	if (previousExecutionFailure != null) {	1✔
63	if (previousExecutionFailure.hasTimeoutFailureInfo() && previousExecutionFailure.hasCause()) {	1✔
64	// It can be only startToClose timeout (others would be fatal).
65	// Structure TIMEOUT_TYPE_START_TO_CLOSE -> TIMEOUT_TYPE_START_TO_CLOSE or
66	// TIMEOUT_TYPE_START_TO_CLOSE -> ApplicationFailure is possible here - chaining of
67	// startToClose timeout with a previous failure.
68	// We reconstruct the last attempt failure (that would typically be preserved in a mutable
69	// state) from local activity execution failure.
70	// See a similar logic in
71	// io.temporal.internal.testservice.StateMachines#timeoutActivityTask.
72	lastAttemptFailure.set(Failure.newBuilder(previousExecutionFailure).clearCause().build());	×
73	} else {
74	lastAttemptFailure.set(previousExecutionFailure);	1✔
75	}
76	}
77	}	1✔
78
79	public String getActivityId() {
80	return executionParams.getActivityId();	1✔
81	}
82
83	public int getCurrentAttempt() {
84	return currentAttempt.get();	1✔
85	}
86
87	/**
88	* The last failure preserved for this activity execution. This field is mimicking the behavior of
89	* {@link PendingActivityInfo#getLastFailure()} that is maintained by the server in the mutable
90	* state and is used to create ActivityFailures with meaningful causes and returned by {@link
91	* io.temporal.api.workflowservice.v1.WorkflowServiceGrpc.WorkflowServiceBlockingStub#describeWorkflowExecution(DescribeWorkflowExecutionRequest)}
92	*/
93	@Nullable
94	public Failure getLastAttemptFailure() {
95	return lastAttemptFailure.get();	1✔
96	}
97
98	@Nullable
99	public Failure getPreviousExecutionFailure() {
100	return executionParams.getPreviousLocalExecutionFailure();	1✔
101	}
102
103	@Nonnull
104	public PollActivityTaskQueueResponse.Builder getInitialTask() {
105	return executionParams.cloneActivityTaskBuilder();	1✔
106	}
107
108	@Nonnull
109	public PollActivityTaskQueueResponse.Builder getNextAttemptActivityTask(
110	@Nullable Failure lastFailure) {
111	// synchronization here is not absolutely needed as LocalActivityWorker#scheduleNextAttempt
112	// shouldn't be executed concurrently. But to make sure this code is safe for future changes,
113	// let's make this method atomic and protect thread-unsafe protobuf builder modification.
114
115	// executionResult here is used just as an internal monitor object that is final and never
116	// escapes the class
117	int nextAttempt;
118	synchronized (executionResult) {	1✔
119	nextAttempt = currentAttempt.incrementAndGet();	1✔
120	if (lastFailure != null) {	1✔
121	this.lastAttemptFailure.set(lastFailure);	1✔
122	}
123	}	1✔
124	// doesn't need to be synchronized as we clone instead of modifying the original task builder
125	return executionParams	1✔
126	.cloneActivityTaskBuilder()	1✔
127	.setAttempt(nextAttempt)	1✔
128	.clearCurrentAttemptScheduledTime();	1✔
129	}
130
131	@Nullable
132	public Deadline getScheduleToCloseDeadline() {
133	return scheduleToCloseDeadline;	1✔
134	}
135
136	public void setScheduleToCloseFuture(@Nullable ScheduledFuture<?> scheduleToCloseFuture) {
137	this.scheduleToCloseFuture = scheduleToCloseFuture;	1✔
138	}	1✔
139
140	public boolean callback(LocalActivityResult result) {
141	if (scheduleToCloseFuture != null) {	1✔
142	scheduleToCloseFuture.cancel(false);	1✔
143	}
144	return executionResult.complete(result);	1✔
145	}
146
147	@Nullable
148	public Duration getScheduleToStartTimeout() {
149	return executionParams.getScheduleToStartTimeout();	1✔
150	}
151
152	public long getOriginalScheduledTimestamp() {
153	return executionParams.getOriginalScheduledTimestamp();	1✔
154	}
155
156	@Nonnull
157	public Duration getLocalRetryThreshold() {
158	return executionParams.getLocalRetryThreshold();	1✔
159	}
160	}

temporalio / sdk-java / #103

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