#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

37.29

/temporal-testing/src/main/java/io/temporal/internal/sync/DummySyncWorkflowContext.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.uber.m3.tally.NoopScope;
import com.uber.m3.tally.Scope;
import io.temporal.api.command.v1.ContinueAsNewWorkflowExecutionCommandAttributes;
import io.temporal.api.command.v1.SignalExternalWorkflowExecutionCommandAttributes;
import io.temporal.api.common.v1.Payload;
import io.temporal.api.common.v1.Payloads;
import io.temporal.api.common.v1.SearchAttributes;
import io.temporal.api.common.v1.WorkflowExecution;
import io.temporal.api.common.v1.WorkflowType;
import io.temporal.api.failure.v1.Failure;
import io.temporal.common.converter.DefaultDataConverter;
import io.temporal.failure.CanceledFailure;
import io.temporal.internal.replay.ReplayWorkflowContext;
import io.temporal.internal.statemachines.ExecuteActivityParameters;
import io.temporal.internal.statemachines.ExecuteLocalActivityParameters;
import io.temporal.internal.statemachines.LocalActivityCallback;
import io.temporal.internal.statemachines.StartChildWorkflowExecutionParameters;
import io.temporal.workflow.Functions;
import java.time.Duration;
import java.util.*;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;

public class DummySyncWorkflowContext {
  public static SyncWorkflowContext newDummySyncWorkflowContext() {
    SyncWorkflowContext context =
        new SyncWorkflowContext(null, DefaultDataConverter.STANDARD_INSTANCE, null);
    context.setReplayContext(new DummyReplayWorkflowContext());
    context.initHeadOutboundCallsInterceptor(context);
    context.initHeadInboundCallsInterceptor(
        new BaseRootWorkflowInboundCallsInterceptor(context) {
          @Override
          public WorkflowOutput execute(WorkflowInput input) {
            throw new UnsupportedOperationException(
                "#execute is not implemented or needed for low level DeterministicRunner tests");
          }
        });
    return context;
  }

  private static final class DummyReplayWorkflowContext implements ReplayWorkflowContext {

    private final Timer timer = new Timer();

    @Override
    public WorkflowExecution getWorkflowExecution() {
      throw new UnsupportedOperationException("not implemented");
    }

    @Override
    public WorkflowExecution getParentWorkflowExecution() {
      throw new UnsupportedOperationException("not implemented");
    }

    @Override
    public WorkflowType getWorkflowType() {
      return WorkflowType.newBuilder().setName("dummy-workflow").build();
    }

    @Override
    public boolean isCancelRequested() {
      throw new UnsupportedOperationException("not implemented");
    }

    @Override
    public ContinueAsNewWorkflowExecutionCommandAttributes getContinueAsNewOnCompletion() {
      throw new UnsupportedOperationException("not implemented");
    }

    @Override
    public Optional<String> getContinuedExecutionRunId() {
      throw new UnsupportedOperationException("not implemented");
    }

    @Override
    public String getTaskQueue() {
      return "dummy-task-queue";
    }

    @Override
    public String getNamespace() {
      return "dummy-namespace";
    }

    @Override
    public String getWorkflowId() {
      return "dummy-workflow-id";
    }

    @Override
    public String getRunId() {
      return "dummy-run-id";
    }

    @Override
    public Duration getWorkflowRunTimeout() {
      throw new UnsupportedOperationException("not implemented");
    }

    @Override
    public Duration getWorkflowExecutionTimeout() {
      return Duration.ZERO;
    }

    @Override
    public long getRunStartedTimestampMillis() {
      return 0;
    }

    @Nonnull
    @Override
    public Duration getWorkflowTaskTimeout() {
      throw new UnsupportedOperationException("not implemented");
    }

    @Override
    public Payload getMemo(String key) {
      throw new UnsupportedOperationException("not implemented");
    }

    @Override
    public SearchAttributes getSearchAttributes() {
      throw new UnsupportedOperationException("not implemented");
    }

    @Override
    public Functions.Proc1<Exception> scheduleActivityTask(
        ExecuteActivityParameters parameters,
        Functions.Proc2<Optional<Payloads>, Failure> callback) {
      throw new UnsupportedOperationException("not implemented");
    }

    @Override
    public Functions.Proc scheduleLocalActivityTask(
        ExecuteLocalActivityParameters parameters, LocalActivityCallback callback) {
      throw new UnsupportedOperationException("not implemented");
    }

    @Override
    public Functions.Proc1<Exception> startChildWorkflow(
        StartChildWorkflowExecutionParameters parameters,
        Functions.Proc2<WorkflowExecution, Exception> executionCallback,
        Functions.Proc2<Optional<Payloads>, Exception> callback) {
      throw new UnsupportedOperationException("not implemented");
    }

    @Override
    public Functions.Proc1<Exception> signalExternalWorkflowExecution(
        SignalExternalWorkflowExecutionCommandAttributes.Builder attributes,
        Functions.Proc2<Void, Failure> callback) {
      throw new UnsupportedOperationException("not implemented");
    }

    @Override
    public void requestCancelExternalWorkflowExecution(
        WorkflowExecution execution, Functions.Proc2<Void, RuntimeException> callback) {
      throw new UnsupportedOperationException("not implemented");
    }

    @Override
    public void continueAsNewOnCompletion(
        ContinueAsNewWorkflowExecutionCommandAttributes attributes) {
      throw new UnsupportedOperationException("not implemented");
    }

    @Override
    public long currentTimeMillis() {
      return System.currentTimeMillis();
    }

    @Override
    public Functions.Proc1<RuntimeException> newTimer(
        Duration delay, Functions.Proc1<RuntimeException> callback) {
      timer.schedule(
          new TimerTask() {
            @Override
            public void run() {
              callback.apply(null);
            }
          },
          delay.toMillis());
      return (e) -> {
        callback.apply(new CanceledFailure(null));
      };
    }

    @Override
    public void sideEffect(
        Functions.Func<Optional<Payloads>> func, Functions.Proc1<Optional<Payloads>> callback) {
      callback.apply(func.apply());
    }

    @Override
    public void mutableSideEffect(
        String id,
        Functions.Func1<Optional<Payloads>, Optional<Payloads>> func,
        Functions.Proc1<Optional<Payloads>> callback) {
      callback.apply(func.apply(Optional.empty()));
    }

    @Override
    public boolean isReplaying() {
      return false;
    }

    @Override
    public void getVersion(
        String changeId,
        int minSupported,
        int maxSupported,
        Functions.Proc2<Integer, RuntimeException> callback) {
      throw new UnsupportedOperationException("not implemented");
    }

    @Override
    public Random newRandom() {
      throw new UnsupportedOperationException("not implemented");
    }

    @Override
    public Scope getMetricsScope() {
      return new NoopScope();
    }

    @Override
    public boolean getEnableLoggingInReplay() {
      return false;
    }

    @Override
    public UUID randomUUID() {
      return UUID.randomUUID();
    }

    @Override
    public void upsertSearchAttributes(SearchAttributes searchAttributes) {
      throw new UnsupportedOperationException("not implemented");
    }

    @Override
    public int getAttempt() {
      return 1;
    }

    @Override
    public String getCronSchedule() {
      return "dummy-cron-schedule";
    }

    @Nullable
    @Override
    public Payloads getLastCompletionResult() {
      return null;
    }

    @Nullable
    @Override
    public Failure getPreviousRunFailure() {
      return null;
    }

    @Nullable
    @Override
    public String getFullReplayDirectQueryName() {
      return null;
    }

    @Override
    public Map<String, Payload> getHeader() {
      return null;
    }

    @Override
    public long getCurrentWorkflowTaskStartedEventId() {
      return 0;
    }
  }
}

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.uber.m3.tally.NoopScope;
24	import com.uber.m3.tally.Scope;
25	import io.temporal.api.command.v1.ContinueAsNewWorkflowExecutionCommandAttributes;
26	import io.temporal.api.command.v1.SignalExternalWorkflowExecutionCommandAttributes;
27	import io.temporal.api.common.v1.Payload;
28	import io.temporal.api.common.v1.Payloads;
29	import io.temporal.api.common.v1.SearchAttributes;
30	import io.temporal.api.common.v1.WorkflowExecution;
31	import io.temporal.api.common.v1.WorkflowType;
32	import io.temporal.api.failure.v1.Failure;
33	import io.temporal.common.converter.DefaultDataConverter;
34	import io.temporal.failure.CanceledFailure;
35	import io.temporal.internal.replay.ReplayWorkflowContext;
36	import io.temporal.internal.statemachines.ExecuteActivityParameters;
37	import io.temporal.internal.statemachines.ExecuteLocalActivityParameters;
38	import io.temporal.internal.statemachines.LocalActivityCallback;
39	import io.temporal.internal.statemachines.StartChildWorkflowExecutionParameters;
40	import io.temporal.workflow.Functions;
41	import java.time.Duration;
42	import java.util.*;
43	import javax.annotation.Nonnull;
44	import javax.annotation.Nullable;
45
46	public class DummySyncWorkflowContext {	×
47	public static SyncWorkflowContext newDummySyncWorkflowContext() {
48	SyncWorkflowContext context =	1✔
49	new SyncWorkflowContext(null, DefaultDataConverter.STANDARD_INSTANCE, null);
50	context.setReplayContext(new DummyReplayWorkflowContext());	1✔
51	context.initHeadOutboundCallsInterceptor(context);	1✔
52	context.initHeadInboundCallsInterceptor(	1✔
53	new BaseRootWorkflowInboundCallsInterceptor(context) {	1✔
54	@Override
55	public WorkflowOutput execute(WorkflowInput input) {
56	throw new UnsupportedOperationException(	×
57	"#execute is not implemented or needed for low level DeterministicRunner tests");
58	}
59	});
60	return context;	1✔
61	}
62
63	private static final class DummyReplayWorkflowContext implements ReplayWorkflowContext {	1✔
64
65	private final Timer timer = new Timer();	1✔
66
67	@Override
68	public WorkflowExecution getWorkflowExecution() {
69	throw new UnsupportedOperationException("not implemented");	×
70	}
71
72	@Override
73	public WorkflowExecution getParentWorkflowExecution() {
74	throw new UnsupportedOperationException("not implemented");	×
75	}
76
77	@Override
78	public WorkflowType getWorkflowType() {
79	return WorkflowType.newBuilder().setName("dummy-workflow").build();	1✔
80	}
81
82	@Override
83	public boolean isCancelRequested() {
84	throw new UnsupportedOperationException("not implemented");	×
85	}
86
87	@Override
88	public ContinueAsNewWorkflowExecutionCommandAttributes getContinueAsNewOnCompletion() {
89	throw new UnsupportedOperationException("not implemented");	×
90	}
91
92	@Override
93	public Optional<String> getContinuedExecutionRunId() {
94	throw new UnsupportedOperationException("not implemented");	×
95	}
96
97	@Override
98	public String getTaskQueue() {
99	return "dummy-task-queue";	1✔
100	}
101
102	@Override
103	public String getNamespace() {
104	return "dummy-namespace";	1✔
105	}
106
107	@Override
108	public String getWorkflowId() {
109	return "dummy-workflow-id";	1✔
110	}
111
112	@Override
113	public String getRunId() {
114	return "dummy-run-id";	1✔
115	}
116
117	@Override
118	public Duration getWorkflowRunTimeout() {
119	throw new UnsupportedOperationException("not implemented");	×
120	}
121
122	@Override
123	public Duration getWorkflowExecutionTimeout() {
124	return Duration.ZERO;	×
125	}
126
127	@Override
128	public long getRunStartedTimestampMillis() {
129	return 0;	×
130	}
131
132	@Nonnull
133	@Override
134	public Duration getWorkflowTaskTimeout() {
135	throw new UnsupportedOperationException("not implemented");	×
136	}
137
138	@Override
139	public Payload getMemo(String key) {
140	throw new UnsupportedOperationException("not implemented");	×
141	}
142
143	@Override
144	public SearchAttributes getSearchAttributes() {
145	throw new UnsupportedOperationException("not implemented");	×
146	}
147
148	@Override
149	public Functions.Proc1<Exception> scheduleActivityTask(
150	ExecuteActivityParameters parameters,
151	Functions.Proc2<Optional<Payloads>, Failure> callback) {
152	throw new UnsupportedOperationException("not implemented");	×
153	}
154
155	@Override
156	public Functions.Proc scheduleLocalActivityTask(
157	ExecuteLocalActivityParameters parameters, LocalActivityCallback callback) {
158	throw new UnsupportedOperationException("not implemented");	×
159	}
160
161	@Override
162	public Functions.Proc1<Exception> startChildWorkflow(
163	StartChildWorkflowExecutionParameters parameters,
164	Functions.Proc2<WorkflowExecution, Exception> executionCallback,
165	Functions.Proc2<Optional<Payloads>, Exception> callback) {
166	throw new UnsupportedOperationException("not implemented");	×
167	}
168
169	@Override
170	public Functions.Proc1<Exception> signalExternalWorkflowExecution(
171	SignalExternalWorkflowExecutionCommandAttributes.Builder attributes,
172	Functions.Proc2<Void, Failure> callback) {
173	throw new UnsupportedOperationException("not implemented");	×
174	}
175
176	@Override
177	public void requestCancelExternalWorkflowExecution(
178	WorkflowExecution execution, Functions.Proc2<Void, RuntimeException> callback) {
179	throw new UnsupportedOperationException("not implemented");	×
180	}
181
182	@Override
183	public void continueAsNewOnCompletion(
184	ContinueAsNewWorkflowExecutionCommandAttributes attributes) {
185	throw new UnsupportedOperationException("not implemented");	×
186	}
187
188	@Override
189	public long currentTimeMillis() {
190	return System.currentTimeMillis();	×
191	}
192
193	@Override
194	public Functions.Proc1<RuntimeException> newTimer(
195	Duration delay, Functions.Proc1<RuntimeException> callback) {
196	timer.schedule(	1✔
197	new TimerTask() {	1✔
198	@Override
199	public void run() {
200	callback.apply(null);	1✔
201	}	1✔
202	},
203	delay.toMillis());	1✔
204	return (e) -> {	1✔
205	callback.apply(new CanceledFailure(null));	1✔
206	};	1✔
207	}
208
209	@Override
210	public void sideEffect(
211	Functions.Func<Optional<Payloads>> func, Functions.Proc1<Optional<Payloads>> callback) {
212	callback.apply(func.apply());	×
213	}	×
214
215	@Override
216	public void mutableSideEffect(
217	String id,
218	Functions.Func1<Optional<Payloads>, Optional<Payloads>> func,
219	Functions.Proc1<Optional<Payloads>> callback) {
220	callback.apply(func.apply(Optional.empty()));	×
221	}	×
222
223	@Override
224	public boolean isReplaying() {
225	return false;	1✔
226	}
227
228	@Override
229	public void getVersion(
230	String changeId,
231	int minSupported,
232	int maxSupported,
233	Functions.Proc2<Integer, RuntimeException> callback) {
234	throw new UnsupportedOperationException("not implemented");	×
235	}
236
237	@Override
238	public Random newRandom() {
239	throw new UnsupportedOperationException("not implemented");	×
240	}
241
242	@Override
243	public Scope getMetricsScope() {
244	return new NoopScope();	×
245	}
246
247	@Override
248	public boolean getEnableLoggingInReplay() {
249	return false;	×
250	}
251
252	@Override
253	public UUID randomUUID() {
254	return UUID.randomUUID();	×
255	}
256
257	@Override
258	public void upsertSearchAttributes(SearchAttributes searchAttributes) {
259	throw new UnsupportedOperationException("not implemented");	×
260	}
261
262	@Override
263	public int getAttempt() {
264	return 1;	×
265	}
266
267	@Override
268	public String getCronSchedule() {
269	return "dummy-cron-schedule";	×
270	}
271
272	@Nullable
273	@Override
274	public Payloads getLastCompletionResult() {
275	return null;	×
276	}
277
278	@Nullable
279	@Override
280	public Failure getPreviousRunFailure() {
281	return null;	×
282	}
283
284	@Nullable
285	@Override
286	public String getFullReplayDirectQueryName() {
287	return null;	×
288	}
289
290	@Override
291	public Map<String, Payload> getHeader() {
292	return null;	×
293	}
294
295	@Override
296	public long getCurrentWorkflowTaskStartedEventId() {
297	return 0;	×
298	}
299	}
300	}

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