#157

pending completion

Build # #157

Build Type

push

github-actions

Committed by

web-flow

Commit Message

Provide SerializationContext for PayloadConverter and PayloadCodec (#1695)

Issue #1694

Run Details

497 of 497 new or added lines in 32 files covered. (100.0%)

16942 of 20806 relevant lines covered (81.43%)

0.81 hits per line

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

89.29

/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncWorkflowWorker.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.temporal.api.common.v1.Payloads;
import io.temporal.common.converter.DataConverter;
import io.temporal.internal.activity.ActivityExecutionContextFactory;
import io.temporal.internal.activity.ActivityTaskHandlerImpl;
import io.temporal.internal.activity.LocalActivityExecutionContextFactoryImpl;
import io.temporal.internal.replay.ReplayWorkflowTaskHandler;
import io.temporal.internal.sync.POJOWorkflowImplementationFactory;
import io.temporal.internal.sync.WorkflowThreadExecutor;
import io.temporal.serviceclient.WorkflowServiceStubs;
import io.temporal.worker.WorkflowImplementationOptions;
import io.temporal.worker.WorkflowTaskDispatchHandle;
import io.temporal.workflow.Functions.Func;
import java.lang.reflect.Type;
import java.time.Duration;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.*;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Facade that supports a lifecycle and maintains an assembly of
 *
 * <ul>
 *   <li>{@link WorkflowWorker} that performing execution of workflow task
 *   <li>{@link LocalActivityWorker} that performs execution of local activities scheduled by the
 *       workflow tasks
 * </ul>
 *
 * and exposing additional management helper methods for the assembly.
 */
public class SyncWorkflowWorker implements SuspendableWorker {
  private static final Logger log = LoggerFactory.getLogger(SyncWorkflowWorker.class);

  private final String identity;
  private final String namespace;
  private final String taskQueue;

  private final WorkflowWorker workflowWorker;
  private final QueryReplayHelper queryReplayHelper;
  private final LocalActivityWorker laWorker;
  private final POJOWorkflowImplementationFactory factory;
  private final DataConverter dataConverter;
  private final ActivityTaskHandlerImpl laTaskHandler;

  public SyncWorkflowWorker(
      @Nonnull WorkflowServiceStubs service,
      @Nonnull String namespace,
      @Nonnull String taskQueue,
      @Nonnull SingleWorkerOptions singleWorkerOptions,
      @Nonnull SingleWorkerOptions localActivityOptions,
      @Nonnull WorkflowRunLockManager runLocks,
      @Nonnull WorkflowExecutorCache cache,
      String stickyTaskQueueName,
      @Nonnull WorkflowThreadExecutor workflowThreadExecutor,
      @Nonnull EagerActivityDispatcher eagerActivityDispatcher) {
    this.identity = singleWorkerOptions.getIdentity();
    this.namespace = namespace;
    this.taskQueue = taskQueue;
    this.dataConverter = singleWorkerOptions.getDataConverter();

    factory =
        new POJOWorkflowImplementationFactory(
            singleWorkerOptions,
            Objects.requireNonNull(workflowThreadExecutor),
            singleWorkerOptions.getWorkerInterceptors(),
            cache,
            namespace);

    ActivityExecutionContextFactory laActivityExecutionContextFactory =
        new LocalActivityExecutionContextFactoryImpl();
    laTaskHandler =
        new ActivityTaskHandlerImpl(
            namespace,
            taskQueue,
            localActivityOptions.getDataConverter(),
            laActivityExecutionContextFactory,
            localActivityOptions.getWorkerInterceptors(),
            localActivityOptions.getContextPropagators());
    laWorker = new LocalActivityWorker(namespace, taskQueue, localActivityOptions, laTaskHandler);

    WorkflowTaskHandler taskHandler =
        new ReplayWorkflowTaskHandler(
            namespace,
            factory,
            cache,
            singleWorkerOptions,
            stickyTaskQueueName,
            singleWorkerOptions.getStickyQueueScheduleToStartTimeout(),
            service,
            laWorker.getLocalActivityScheduler());

    workflowWorker =
        new WorkflowWorker(
            service,
            namespace,
            taskQueue,
            stickyTaskQueueName,
            singleWorkerOptions,
            runLocks,
            cache,
            taskHandler,
            eagerActivityDispatcher);

    // Exists to support Worker#replayWorkflowExecution functionality.
    // This handler has to be non-sticky to avoid evicting actual executions from the cache
    WorkflowTaskHandler nonStickyReplayTaskHandler =
        new ReplayWorkflowTaskHandler(
            namespace,
            factory,
            null,
            singleWorkerOptions,
            null,
            Duration.ZERO,
            service,
            laWorker.getLocalActivityScheduler());

    queryReplayHelper = new QueryReplayHelper(nonStickyReplayTaskHandler);
  }

  public void registerWorkflowImplementationTypes(
      WorkflowImplementationOptions options, Class<?>[] workflowImplementationTypes) {
    factory.registerWorkflowImplementationTypes(options, workflowImplementationTypes);
  }

  public <R> void registerWorkflowImplementationFactory(
      WorkflowImplementationOptions options, Class<R> clazz, Func<R> factory) {
    this.factory.addWorkflowImplementationFactory(options, clazz, factory);
  }

  public void registerLocalActivityImplementations(Object... activitiesImplementation) {
    this.laTaskHandler.registerActivityImplementations(activitiesImplementation);
  }

  @Override
  public boolean start() {
    boolean started = workflowWorker.start();
    // It doesn't start if no types are registered with it.
    if (started) {
      laWorker.start();
    }
    return started;
  }

  @Override
  public CompletableFuture<Void> shutdown(ShutdownManager shutdownManager, boolean interruptTasks) {
    return workflowWorker
        .shutdown(shutdownManager, interruptTasks)
        .thenCompose(ignore -> laWorker.shutdown(shutdownManager, interruptTasks))
        .exceptionally(
            e -> {
              log.error("[BUG] Unexpected exception during shutdown", e);
              return null;
            });
  }

  @Override
  public void awaitTermination(long timeout, TimeUnit unit) {
    long timeoutMillis = ShutdownManager.awaitTermination(laWorker, unit.toMillis(timeout));
    ShutdownManager.awaitTermination(workflowWorker, timeoutMillis);
  }

  @Override
  public void suspendPolling() {
    workflowWorker.suspendPolling();
  }

  @Override
  public void resumePolling() {
    workflowWorker.resumePolling();
  }

  @SuppressWarnings("deprecation")
  public <R> R queryWorkflowExecution(
      io.temporal.internal.common.WorkflowExecutionHistory history,
      String queryType,
      Class<R> resultClass,
      Type resultType,
      Object[] args)
      throws Exception {
    Optional<Payloads> serializedArgs = dataConverter.toPayloads(args);
    Optional<Payloads> result =
        queryReplayHelper.queryWorkflowExecution(history, queryType, serializedArgs);
    return dataConverter.fromPayloads(0, result, resultClass, resultType);
  }

  @Override
  public boolean isSuspended() {
    return workflowWorker.isSuspended();
  }

  @Override
  public boolean isShutdown() {
    return workflowWorker.isShutdown() || laWorker.isShutdown();
  }

  @Override
  public boolean isTerminated() {
    return workflowWorker.isTerminated() && laWorker.isTerminated();
  }

  @Override
  public WorkerLifecycleState getLifecycleState() {
    return null;
  }

  @Override
  public String toString() {
    return String.format(
        "SyncWorkflowWorker{namespace=%s, taskQueue=%s, identity=%s}",
        namespace, taskQueue, identity);
  }

  @Nullable
  public WorkflowTaskDispatchHandle reserveWorkflowExecutor() {
    return workflowWorker.reserveWorkflowExecutor();
  }
}

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.temporal.api.common.v1.Payloads;
24	import io.temporal.common.converter.DataConverter;
25	import io.temporal.internal.activity.ActivityExecutionContextFactory;
26	import io.temporal.internal.activity.ActivityTaskHandlerImpl;
27	import io.temporal.internal.activity.LocalActivityExecutionContextFactoryImpl;
28	import io.temporal.internal.replay.ReplayWorkflowTaskHandler;
29	import io.temporal.internal.sync.POJOWorkflowImplementationFactory;
30	import io.temporal.internal.sync.WorkflowThreadExecutor;
31	import io.temporal.serviceclient.WorkflowServiceStubs;
32	import io.temporal.worker.WorkflowImplementationOptions;
33	import io.temporal.worker.WorkflowTaskDispatchHandle;
34	import io.temporal.workflow.Functions.Func;
35	import java.lang.reflect.Type;
36	import java.time.Duration;
37	import java.util.Objects;
38	import java.util.Optional;
39	import java.util.concurrent.*;
40	import javax.annotation.Nonnull;
41	import javax.annotation.Nullable;
42	import org.slf4j.Logger;
43	import org.slf4j.LoggerFactory;
44
45	/**
46	* Facade that supports a lifecycle and maintains an assembly of
47	*
48	* <ul>
49	* <li>{@link WorkflowWorker} that performing execution of workflow task
50	* <li>{@link LocalActivityWorker} that performs execution of local activities scheduled by the
51	* workflow tasks
52	* </ul>
53	*
54	* and exposing additional management helper methods for the assembly.
55	*/
56	public class SyncWorkflowWorker implements SuspendableWorker {
57	private static final Logger log = LoggerFactory.getLogger(SyncWorkflowWorker.class);
58
59	private final String identity;
60	private final String namespace;
61	private final String taskQueue;
62
63	private final WorkflowWorker workflowWorker;
64	private final QueryReplayHelper queryReplayHelper;
65	private final LocalActivityWorker laWorker;
66	private final POJOWorkflowImplementationFactory factory;
67	private final DataConverter dataConverter;
68	private final ActivityTaskHandlerImpl laTaskHandler;
69
70	public SyncWorkflowWorker(
71	@Nonnull WorkflowServiceStubs service,
72	@Nonnull String namespace,
73	@Nonnull String taskQueue,
74	@Nonnull SingleWorkerOptions singleWorkerOptions,
75	@Nonnull SingleWorkerOptions localActivityOptions,
76	@Nonnull WorkflowRunLockManager runLocks,
77	@Nonnull WorkflowExecutorCache cache,
78	String stickyTaskQueueName,
79	@Nonnull WorkflowThreadExecutor workflowThreadExecutor,
80	@Nonnull EagerActivityDispatcher eagerActivityDispatcher) {
81	this.identity = singleWorkerOptions.getIdentity();
82	this.namespace = namespace;
83	this.taskQueue = taskQueue;
84	this.dataConverter = singleWorkerOptions.getDataConverter();
85
86	factory =
87	new POJOWorkflowImplementationFactory(
88	singleWorkerOptions,
89	Objects.requireNonNull(workflowThreadExecutor),
90	singleWorkerOptions.getWorkerInterceptors(),
91	cache,
92	namespace);
93
94	ActivityExecutionContextFactory laActivityExecutionContextFactory =
95	new LocalActivityExecutionContextFactoryImpl();
96	laTaskHandler =
97	new ActivityTaskHandlerImpl(
98	namespace,
99	taskQueue,
100	localActivityOptions.getDataConverter(),
101	laActivityExecutionContextFactory,
102	localActivityOptions.getWorkerInterceptors(),
103	localActivityOptions.getContextPropagators());
104	laWorker = new LocalActivityWorker(namespace, taskQueue, localActivityOptions, laTaskHandler);
105
106	WorkflowTaskHandler taskHandler =
107	new ReplayWorkflowTaskHandler(
108	namespace,
109	factory,
110	cache,
111	singleWorkerOptions,
112	stickyTaskQueueName,
113	singleWorkerOptions.getStickyQueueScheduleToStartTimeout(),
114	service,
115	laWorker.getLocalActivityScheduler());
116
117	workflowWorker =
118	new WorkflowWorker(
119	service,
120	namespace,
121	taskQueue,
122	stickyTaskQueueName,
123	singleWorkerOptions,
124	runLocks,
125	cache,
126	taskHandler,
127	eagerActivityDispatcher);
128
129	// Exists to support Worker#replayWorkflowExecution functionality.
130	// This handler has to be non-sticky to avoid evicting actual executions from the cache
131	WorkflowTaskHandler nonStickyReplayTaskHandler =
132	new ReplayWorkflowTaskHandler(
133	namespace,
134	factory,
135	null,
136	singleWorkerOptions,
137	null,
138	Duration.ZERO,
139	service,
140	laWorker.getLocalActivityScheduler());
141
142	queryReplayHelper = new QueryReplayHelper(nonStickyReplayTaskHandler);
143	}
144
145	public void registerWorkflowImplementationTypes(
146	WorkflowImplementationOptions options, Class<?>[] workflowImplementationTypes) {
147	factory.registerWorkflowImplementationTypes(options, workflowImplementationTypes);
148	}
149
150	public <R> void registerWorkflowImplementationFactory(
151	WorkflowImplementationOptions options, Class<R> clazz, Func<R> factory) {
152	this.factory.addWorkflowImplementationFactory(options, clazz, factory);
153	}
154
155	public void registerLocalActivityImplementations(Object... activitiesImplementation) {
156	this.laTaskHandler.registerActivityImplementations(activitiesImplementation);
157	}
158
159	@Override
160	public boolean start() {
161	boolean started = workflowWorker.start();
162	// It doesn't start if no types are registered with it.
163	if (started) {
164	laWorker.start();
165	}
166	return started;
167	}
168
169	@Override
170	public CompletableFuture<Void> shutdown(ShutdownManager shutdownManager, boolean interruptTasks) {
171	return workflowWorker
172	.shutdown(shutdownManager, interruptTasks)
173	.thenCompose(ignore -> laWorker.shutdown(shutdownManager, interruptTasks))
174	.exceptionally(
175	e -> {
176	log.error("[BUG] Unexpected exception during shutdown", e);
177	return null;
178	});
179	}
180
181	@Override
182	public void awaitTermination(long timeout, TimeUnit unit) {
183	long timeoutMillis = ShutdownManager.awaitTermination(laWorker, unit.toMillis(timeout));
184	ShutdownManager.awaitTermination(workflowWorker, timeoutMillis);
185	}
186
187	@Override
188	public void suspendPolling() {
189	workflowWorker.suspendPolling();
190	}
191
192	@Override
193	public void resumePolling() {
194	workflowWorker.resumePolling();
195	}
196
197	@SuppressWarnings("deprecation")
198	public <R> R queryWorkflowExecution(
199	io.temporal.internal.common.WorkflowExecutionHistory history,
200	String queryType,
201	Class<R> resultClass,
202	Type resultType,
203	Object[] args)
204	throws Exception {
205	Optional<Payloads> serializedArgs = dataConverter.toPayloads(args);
206	Optional<Payloads> result =
207	queryReplayHelper.queryWorkflowExecution(history, queryType, serializedArgs);
208	return dataConverter.fromPayloads(0, result, resultClass, resultType);
209	}
210
211	@Override
212	public boolean isSuspended() {
213	return workflowWorker.isSuspended();
214	}
215
216	@Override
217	public boolean isShutdown() {
218	return workflowWorker.isShutdown() \|\| laWorker.isShutdown();
219	}
220
221	@Override
222	public boolean isTerminated() {
223	return workflowWorker.isTerminated() && laWorker.isTerminated();
224	}
225
226	@Override
227	public WorkerLifecycleState getLifecycleState() {
228	return null;
229	}
230
231	@Override
232	public String toString() {
233	return String.format(
234	"SyncWorkflowWorker{namespace=%s, taskQueue=%s, identity=%s}",
235	namespace, taskQueue, identity);
236	}
237
238	@Nullable
239	public WorkflowTaskDispatchHandle reserveWorkflowExecutor() {
240	return workflowWorker.reserveWorkflowExecutor();
241	}
242	}

temporalio / sdk-java / #157

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