#175

pending completion

Build # #175

Build Type

push

github-actions

Committed by

web-flow

Commit Message

Worker / Build Id versioning (#1786)

Implement new worker build id based versioning feature

Run Details

236 of 236 new or added lines in 24 files covered. (100.0%)

18343 of 23697 relevant lines covered (77.41%)

0.81 hits per line

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

75.25

/temporal-sdk/src/main/java/io/temporal/internal/worker/SingleWorkerOptions.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 com.uber.m3.tally.NoopScope;
import com.uber.m3.tally.Scope;
import io.temporal.api.common.v1.WorkerVersionStamp;
import io.temporal.common.context.ContextPropagator;
import io.temporal.common.converter.DataConverter;
import io.temporal.common.converter.GlobalDataConverter;
import io.temporal.common.interceptors.WorkerInterceptor;
import java.time.Duration;
import java.util.List;

public final class SingleWorkerOptions {

  public static Builder newBuilder() {
    return new Builder();
  }

  public static Builder newBuilder(SingleWorkerOptions options) {
    return new Builder(options);
  }

  public static final class Builder {

    private String identity;
    private String binaryChecksum;
    private String buildId;
    private boolean useBuildIdForVersioning;
    private DataConverter dataConverter;
    private int taskExecutorThreadPoolSize = 100;
    private PollerOptions pollerOptions;
    private Scope metricsScope;
    private boolean enableLoggingInReplay;
    private List<ContextPropagator> contextPropagators;
    private WorkerInterceptor[] workerInterceptors;

    private Duration stickyQueueScheduleToStartTimeout;
    private long defaultDeadlockDetectionTimeout;
    private Duration maxHeartbeatThrottleInterval;
    private Duration defaultHeartbeatThrottleInterval;

    private Builder() {}

    private Builder(SingleWorkerOptions options) {
      if (options == null) {
        return;
      }
      this.identity = options.getIdentity();
      this.binaryChecksum = options.getBinaryChecksum();
      this.dataConverter = options.getDataConverter();
      this.pollerOptions = options.getPollerOptions();
      this.taskExecutorThreadPoolSize = options.getTaskExecutorThreadPoolSize();
      this.metricsScope = options.getMetricsScope();
      this.enableLoggingInReplay = options.getEnableLoggingInReplay();
      this.contextPropagators = options.getContextPropagators();
      this.workerInterceptors = options.getWorkerInterceptors();
      this.stickyQueueScheduleToStartTimeout = options.getStickyQueueScheduleToStartTimeout();
      this.defaultDeadlockDetectionTimeout = options.getDefaultDeadlockDetectionTimeout();
      this.maxHeartbeatThrottleInterval = options.getMaxHeartbeatThrottleInterval();
      this.defaultHeartbeatThrottleInterval = options.getDefaultHeartbeatThrottleInterval();
      this.buildId = options.getBuildId();
      this.useBuildIdForVersioning = options.isUsingBuildIdForVersioning();
    }

    public Builder setIdentity(String identity) {
      this.identity = identity;
      return this;
    }

    @Deprecated
    public Builder setBinaryChecksum(String binaryChecksum) {
      this.binaryChecksum = binaryChecksum;
      return this;
    }

    public Builder setDataConverter(DataConverter dataConverter) {
      this.dataConverter = dataConverter;
      return this;
    }

    public Builder setTaskExecutorThreadPoolSize(int taskExecutorThreadPoolSize) {
      this.taskExecutorThreadPoolSize = taskExecutorThreadPoolSize;
      return this;
    }

    public Builder setPollerOptions(PollerOptions pollerOptions) {
      this.pollerOptions = pollerOptions;
      return this;
    }

    public Builder setMetricsScope(Scope metricsScope) {
      this.metricsScope = metricsScope;
      return this;
    }

    public Builder setEnableLoggingInReplay(boolean enableLoggingInReplay) {
      this.enableLoggingInReplay = enableLoggingInReplay;
      return this;
    }

    /** Specifies the list of context propagators to use during this workflow. */
    public Builder setContextPropagators(List<ContextPropagator> contextPropagators) {
      this.contextPropagators = contextPropagators;
      return this;
    }

    /** Specifies the list of worker interceptors to use during this workflow. */
    public Builder setWorkerInterceptors(WorkerInterceptor[] workerInterceptors) {
      this.workerInterceptors = workerInterceptors;
      return this;
    }

    public Builder setStickyQueueScheduleToStartTimeout(
        Duration stickyQueueScheduleToStartTimeout) {
      this.stickyQueueScheduleToStartTimeout = stickyQueueScheduleToStartTimeout;
      return this;
    }

    public Builder setDefaultDeadlockDetectionTimeout(long defaultDeadlockDetectionTimeout) {
      this.defaultDeadlockDetectionTimeout = defaultDeadlockDetectionTimeout;
      return this;
    }

    public Builder setMaxHeartbeatThrottleInterval(Duration maxHeartbeatThrottleInterval) {
      this.maxHeartbeatThrottleInterval = maxHeartbeatThrottleInterval;
      return this;
    }

    public Builder setDefaultHeartbeatThrottleInterval(Duration defaultHeartbeatThrottleInterval) {
      this.defaultHeartbeatThrottleInterval = defaultHeartbeatThrottleInterval;
      return this;
    }

    public Builder setBuildId(String buildId) {
      this.buildId = buildId;
      return this;
    }

    public Builder setUseBuildIdForVersioning(boolean useBuildIdForVersioning) {
      this.useBuildIdForVersioning = useBuildIdForVersioning;
      return this;
    }

    public SingleWorkerOptions build() {
      PollerOptions pollerOptions = this.pollerOptions;
      if (pollerOptions == null) {
        pollerOptions = PollerOptions.newBuilder().build();
      }

      DataConverter dataConverter = this.dataConverter;
      if (dataConverter == null) {
        dataConverter = GlobalDataConverter.get();
      }

      Scope metricsScope = this.metricsScope;
      if (metricsScope == null) {
        metricsScope = new NoopScope();
      }

      return new SingleWorkerOptions(
          this.identity,
          this.binaryChecksum,
          this.buildId,
          this.useBuildIdForVersioning,
          dataConverter,
          this.taskExecutorThreadPoolSize,
          pollerOptions,
          metricsScope,
          this.enableLoggingInReplay,
          this.contextPropagators,
          this.workerInterceptors,
          this.stickyQueueScheduleToStartTimeout,
          this.defaultDeadlockDetectionTimeout,
          this.maxHeartbeatThrottleInterval,
          this.defaultHeartbeatThrottleInterval);
    }
  }

  private final String identity;
  private final String binaryChecksum;
  private final String buildId;
  private final boolean useBuildIdForVersioning;
  private final DataConverter dataConverter;
  private final int taskExecutorThreadPoolSize;
  private final PollerOptions pollerOptions;
  private final Scope metricsScope;
  private final boolean enableLoggingInReplay;
  private final List<ContextPropagator> contextPropagators;
  private final WorkerInterceptor[] workerInterceptors;
  private final Duration stickyQueueScheduleToStartTimeout;
  private final long defaultDeadlockDetectionTimeout;
  private final Duration maxHeartbeatThrottleInterval;
  private final Duration defaultHeartbeatThrottleInterval;

  private SingleWorkerOptions(
      String identity,
      String binaryChecksum,
      String buildId,
      boolean useBuildIdForVersioning,
      DataConverter dataConverter,
      int taskExecutorThreadPoolSize,
      PollerOptions pollerOptions,
      Scope metricsScope,
      boolean enableLoggingInReplay,
      List<ContextPropagator> contextPropagators,
      WorkerInterceptor[] workerInterceptors,
      Duration stickyQueueScheduleToStartTimeout,
      long defaultDeadlockDetectionTimeout,
      Duration maxHeartbeatThrottleInterval,
      Duration defaultHeartbeatThrottleInterval) {
    this.identity = identity;
    this.binaryChecksum = binaryChecksum;
    this.buildId = buildId;
    this.useBuildIdForVersioning = useBuildIdForVersioning;
    this.dataConverter = dataConverter;
    this.taskExecutorThreadPoolSize = taskExecutorThreadPoolSize;
    this.pollerOptions = pollerOptions;
    this.metricsScope = metricsScope;
    this.enableLoggingInReplay = enableLoggingInReplay;
    this.contextPropagators = contextPropagators;
    this.workerInterceptors = workerInterceptors;
    this.stickyQueueScheduleToStartTimeout = stickyQueueScheduleToStartTimeout;
    this.defaultDeadlockDetectionTimeout = defaultDeadlockDetectionTimeout;
    this.maxHeartbeatThrottleInterval = maxHeartbeatThrottleInterval;
    this.defaultHeartbeatThrottleInterval = defaultHeartbeatThrottleInterval;
  }

  public String getIdentity() {
    return identity;
  }

  @Deprecated
  public String getBinaryChecksum() {
    return binaryChecksum;
  }

  public String getBuildId() {
    if (buildId == null) {
      return binaryChecksum;
    }
    return buildId;
  }

  public boolean isUsingBuildIdForVersioning() {
    return useBuildIdForVersioning;
  }

  public DataConverter getDataConverter() {
    return dataConverter;
  }

  public int getTaskExecutorThreadPoolSize() {
    return taskExecutorThreadPoolSize;
  }

  public PollerOptions getPollerOptions() {
    return pollerOptions;
  }

  public Scope getMetricsScope() {
    return metricsScope;
  }

  public boolean getEnableLoggingInReplay() {
    return enableLoggingInReplay;
  }

  public List<ContextPropagator> getContextPropagators() {
    return contextPropagators;
  }

  public WorkerInterceptor[] getWorkerInterceptors() {
    return workerInterceptors;
  }

  public Duration getStickyQueueScheduleToStartTimeout() {
    return stickyQueueScheduleToStartTimeout;
  }

  public long getDefaultDeadlockDetectionTimeout() {
    return defaultDeadlockDetectionTimeout;
  }

  public Duration getMaxHeartbeatThrottleInterval() {
    return maxHeartbeatThrottleInterval;
  }

  public Duration getDefaultHeartbeatThrottleInterval() {
    return defaultHeartbeatThrottleInterval;
  }

  public WorkerVersionStamp workerVersionStamp() {
    return WorkerVersionStamp.newBuilder()
        .setBuildId(this.getBuildId())
        .setUseVersioning(this.isUsingBuildIdForVersioning())
        .build();
  }
}

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 com.uber.m3.tally.NoopScope;
24	import com.uber.m3.tally.Scope;
25	import io.temporal.api.common.v1.WorkerVersionStamp;
26	import io.temporal.common.context.ContextPropagator;
27	import io.temporal.common.converter.DataConverter;
28	import io.temporal.common.converter.GlobalDataConverter;
29	import io.temporal.common.interceptors.WorkerInterceptor;
30	import java.time.Duration;
31	import java.util.List;
32
33	public final class SingleWorkerOptions {
34
35	public static Builder newBuilder() {
36	return new Builder();	1✔
37	}
38
39	public static Builder newBuilder(SingleWorkerOptions options) {
40	return new Builder(options);	×
41	}
42
43	public static final class Builder {
44
45	private String identity;
46	private String binaryChecksum;
47	private String buildId;
48	private boolean useBuildIdForVersioning;
49	private DataConverter dataConverter;
50	private int taskExecutorThreadPoolSize = 100;	1✔
51	private PollerOptions pollerOptions;
52	private Scope metricsScope;
53	private boolean enableLoggingInReplay;
54	private List<ContextPropagator> contextPropagators;
55	private WorkerInterceptor[] workerInterceptors;
56
57	private Duration stickyQueueScheduleToStartTimeout;
58	private long defaultDeadlockDetectionTimeout;
59	private Duration maxHeartbeatThrottleInterval;
60	private Duration defaultHeartbeatThrottleInterval;
61
62	private Builder() {}	1✔
63
64	private Builder(SingleWorkerOptions options) {	×
65	if (options == null) {	×
66	return;	×
67	}
68	this.identity = options.getIdentity();	×
69	this.binaryChecksum = options.getBinaryChecksum();	×
70	this.dataConverter = options.getDataConverter();	×
71	this.pollerOptions = options.getPollerOptions();	×
72	this.taskExecutorThreadPoolSize = options.getTaskExecutorThreadPoolSize();	×
73	this.metricsScope = options.getMetricsScope();	×
74	this.enableLoggingInReplay = options.getEnableLoggingInReplay();	×
75	this.contextPropagators = options.getContextPropagators();	×
76	this.workerInterceptors = options.getWorkerInterceptors();	×
77	this.stickyQueueScheduleToStartTimeout = options.getStickyQueueScheduleToStartTimeout();	×
78	this.defaultDeadlockDetectionTimeout = options.getDefaultDeadlockDetectionTimeout();	×
79	this.maxHeartbeatThrottleInterval = options.getMaxHeartbeatThrottleInterval();	×
80	this.defaultHeartbeatThrottleInterval = options.getDefaultHeartbeatThrottleInterval();	×
81	this.buildId = options.getBuildId();	×
82	this.useBuildIdForVersioning = options.isUsingBuildIdForVersioning();	×
83	}	×
84
85	public Builder setIdentity(String identity) {
86	this.identity = identity;	1✔
87	return this;	1✔
88	}
89
90	@Deprecated
91	public Builder setBinaryChecksum(String binaryChecksum) {
92	this.binaryChecksum = binaryChecksum;	×
93	return this;	×
94	}
95
96	public Builder setDataConverter(DataConverter dataConverter) {
97	this.dataConverter = dataConverter;	1✔
98	return this;	1✔
99	}
100
101	public Builder setTaskExecutorThreadPoolSize(int taskExecutorThreadPoolSize) {
102	this.taskExecutorThreadPoolSize = taskExecutorThreadPoolSize;	1✔
103	return this;	1✔
104	}
105
106	public Builder setPollerOptions(PollerOptions pollerOptions) {
107	this.pollerOptions = pollerOptions;	1✔
108	return this;	1✔
109	}
110
111	public Builder setMetricsScope(Scope metricsScope) {
112	this.metricsScope = metricsScope;	1✔
113	return this;	1✔
114	}
115
116	public Builder setEnableLoggingInReplay(boolean enableLoggingInReplay) {
117	this.enableLoggingInReplay = enableLoggingInReplay;	1✔
118	return this;	1✔
119	}
120
121	/** Specifies the list of context propagators to use during this workflow. */
122	public Builder setContextPropagators(List<ContextPropagator> contextPropagators) {
123	this.contextPropagators = contextPropagators;	1✔
124	return this;	1✔
125	}
126
127	/** Specifies the list of worker interceptors to use during this workflow. */
128	public Builder setWorkerInterceptors(WorkerInterceptor[] workerInterceptors) {
129	this.workerInterceptors = workerInterceptors;	1✔
130	return this;	1✔
131	}
132
133	public Builder setStickyQueueScheduleToStartTimeout(
134	Duration stickyQueueScheduleToStartTimeout) {
135	this.stickyQueueScheduleToStartTimeout = stickyQueueScheduleToStartTimeout;	1✔
136	return this;	1✔
137	}
138
139	public Builder setDefaultDeadlockDetectionTimeout(long defaultDeadlockDetectionTimeout) {
140	this.defaultDeadlockDetectionTimeout = defaultDeadlockDetectionTimeout;	1✔
141	return this;	1✔
142	}
143
144	public Builder setMaxHeartbeatThrottleInterval(Duration maxHeartbeatThrottleInterval) {
145	this.maxHeartbeatThrottleInterval = maxHeartbeatThrottleInterval;	1✔
146	return this;	1✔
147	}
148
149	public Builder setDefaultHeartbeatThrottleInterval(Duration defaultHeartbeatThrottleInterval) {
150	this.defaultHeartbeatThrottleInterval = defaultHeartbeatThrottleInterval;	1✔
151	return this;	1✔
152	}
153
154	public Builder setBuildId(String buildId) {
155	this.buildId = buildId;	1✔
156	return this;	1✔
157	}
158
159	public Builder setUseBuildIdForVersioning(boolean useBuildIdForVersioning) {
160	this.useBuildIdForVersioning = useBuildIdForVersioning;	1✔
161	return this;	1✔
162	}
163
164	public SingleWorkerOptions build() {
165	PollerOptions pollerOptions = this.pollerOptions;	1✔
166	if (pollerOptions == null) {	1✔
167	pollerOptions = PollerOptions.newBuilder().build();	1✔
168	}
169
170	DataConverter dataConverter = this.dataConverter;	1✔
171	if (dataConverter == null) {	1✔
172	dataConverter = GlobalDataConverter.get();	1✔
173	}
174
175	Scope metricsScope = this.metricsScope;	1✔
176	if (metricsScope == null) {	1✔
177	metricsScope = new NoopScope();	1✔
178	}
179
180	return new SingleWorkerOptions(	1✔
181	this.identity,
182	this.binaryChecksum,
183	this.buildId,
184	this.useBuildIdForVersioning,
185	dataConverter,
186	this.taskExecutorThreadPoolSize,
187	pollerOptions,
188	metricsScope,
189	this.enableLoggingInReplay,
190	this.contextPropagators,
191	this.workerInterceptors,
192	this.stickyQueueScheduleToStartTimeout,
193	this.defaultDeadlockDetectionTimeout,
194	this.maxHeartbeatThrottleInterval,
195	this.defaultHeartbeatThrottleInterval);
196	}
197	}
198
199	private final String identity;
200	private final String binaryChecksum;
201	private final String buildId;
202	private final boolean useBuildIdForVersioning;
203	private final DataConverter dataConverter;
204	private final int taskExecutorThreadPoolSize;
205	private final PollerOptions pollerOptions;
206	private final Scope metricsScope;
207	private final boolean enableLoggingInReplay;
208	private final List<ContextPropagator> contextPropagators;
209	private final WorkerInterceptor[] workerInterceptors;
210	private final Duration stickyQueueScheduleToStartTimeout;
211	private final long defaultDeadlockDetectionTimeout;
212	private final Duration maxHeartbeatThrottleInterval;
213	private final Duration defaultHeartbeatThrottleInterval;
214
215	private SingleWorkerOptions(
216	String identity,
217	String binaryChecksum,
218	String buildId,
219	boolean useBuildIdForVersioning,
220	DataConverter dataConverter,
221	int taskExecutorThreadPoolSize,
222	PollerOptions pollerOptions,
223	Scope metricsScope,
224	boolean enableLoggingInReplay,
225	List<ContextPropagator> contextPropagators,
226	WorkerInterceptor[] workerInterceptors,
227	Duration stickyQueueScheduleToStartTimeout,
228	long defaultDeadlockDetectionTimeout,
229	Duration maxHeartbeatThrottleInterval,
230	Duration defaultHeartbeatThrottleInterval) {	1✔
231	this.identity = identity;	1✔
232	this.binaryChecksum = binaryChecksum;	1✔
233	this.buildId = buildId;	1✔
234	this.useBuildIdForVersioning = useBuildIdForVersioning;	1✔
235	this.dataConverter = dataConverter;	1✔
236	this.taskExecutorThreadPoolSize = taskExecutorThreadPoolSize;	1✔
237	this.pollerOptions = pollerOptions;	1✔
238	this.metricsScope = metricsScope;	1✔
239	this.enableLoggingInReplay = enableLoggingInReplay;	1✔
240	this.contextPropagators = contextPropagators;	1✔
241	this.workerInterceptors = workerInterceptors;	1✔
242	this.stickyQueueScheduleToStartTimeout = stickyQueueScheduleToStartTimeout;	1✔
243	this.defaultDeadlockDetectionTimeout = defaultDeadlockDetectionTimeout;	1✔
244	this.maxHeartbeatThrottleInterval = maxHeartbeatThrottleInterval;	1✔
245	this.defaultHeartbeatThrottleInterval = defaultHeartbeatThrottleInterval;	1✔
246	}	1✔
247
248	public String getIdentity() {
249	return identity;	1✔
250	}
251
252	@Deprecated
253	public String getBinaryChecksum() {
254	return binaryChecksum;	×
255	}
256
257	public String getBuildId() {
258	if (buildId == null) {	1✔
259	return binaryChecksum;	×
260	}
261	return buildId;	1✔
262	}
263
264	public boolean isUsingBuildIdForVersioning() {
265	return useBuildIdForVersioning;	1✔
266	}
267
268	public DataConverter getDataConverter() {
269	return dataConverter;	1✔
270	}
271
272	public int getTaskExecutorThreadPoolSize() {
273	return taskExecutorThreadPoolSize;	1✔
274	}
275
276	public PollerOptions getPollerOptions() {
277	return pollerOptions;	1✔
278	}
279
280	public Scope getMetricsScope() {
281	return metricsScope;	1✔
282	}
283
284	public boolean getEnableLoggingInReplay() {
285	return enableLoggingInReplay;	×
286	}
287
288	public List<ContextPropagator> getContextPropagators() {
289	return contextPropagators;	1✔
290	}
291
292	public WorkerInterceptor[] getWorkerInterceptors() {
293	return workerInterceptors;	1✔
294	}
295
296	public Duration getStickyQueueScheduleToStartTimeout() {
297	return stickyQueueScheduleToStartTimeout;	1✔
298	}
299
300	public long getDefaultDeadlockDetectionTimeout() {
301	return defaultDeadlockDetectionTimeout;	1✔
302	}
303
304	public Duration getMaxHeartbeatThrottleInterval() {
305	return maxHeartbeatThrottleInterval;	1✔
306	}
307
308	public Duration getDefaultHeartbeatThrottleInterval() {
309	return defaultHeartbeatThrottleInterval;	1✔
310	}
311
312	public WorkerVersionStamp workerVersionStamp() {
313	return WorkerVersionStamp.newBuilder()	1✔
314	.setBuildId(this.getBuildId())	1✔
315	.setUseVersioning(this.isUsingBuildIdForVersioning())	1✔
316	.build();	1✔
317	}
318	}

temporalio / sdk-java / #175

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