2428

Committed 11 Jul 2024 09:55PM UTC coverage: 61.938% (+0.5%) from 61.478%

Build # 2428

Build Type

push

buildkite

Committed by

web-flow

Commit Message

Use the same Request ID for retries (#915)

Currently the RequestId is initialized in the implementation of IWorkflowService (either Thrift2ProtoAdapter or WorkflowServiceTChannel). Since retries are handled within GenericWorkflowClientExternalImpl, each retry uses a different request ID. This causes Cadence's idempotency feature not to work correctly for retries from the Java client.

Move Request ID initialization to GenericWorkflowClientExternalImpl, and for backwards compatibility (since users may be interacting with the IWorkflowService directly) continue setting the request ID if not specified.

Run Details

14 of 27 new or added lines in 3 files covered. (51.85%)

13 existing lines in 5 files now uncovered.

12071 of 19489 relevant lines covered (61.94%)

0.62 hits per line

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

92.65

/src/main/java/com/uber/cadence/internal/sync/CancellationScopeImpl.java

/*
 *  Copyright 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 file except in compliance with the License. A copy of the License is
 *  located at
 *
 *  http://aws.amazon.com/apache2.0
 *
 *  or in the "license" file accompanying this file. This file 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 com.uber.cadence.internal.sync;

import com.uber.cadence.workflow.CancellationScope;
import com.uber.cadence.workflow.CompletablePromise;
import com.uber.cadence.workflow.Functions;
import com.uber.cadence.workflow.Workflow;
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.Set;

class CancellationScopeImpl implements CancellationScope {

  private static ThreadLocal<Deque<CancellationScopeImpl>> scopeStack =
      ThreadLocal.withInitial(ArrayDeque::new);
  private boolean detached;
  private CompletablePromise<String> cancellationPromise;

  static CancellationScopeImpl current() {
    if (scopeStack.get().isEmpty()) {
      throw new IllegalStateException("Cannot be called by non workflow thread");
    }
    return scopeStack.get().peekFirst();
  }

  private static void pushCurrent(CancellationScopeImpl scope) {
    scopeStack.get().addFirst(scope);
  }

  private static void popCurrent(CancellationScopeImpl expected) {
    CancellationScopeImpl current = scopeStack.get().pollFirst();
    if (current != expected) {
      throw new Error("Unexpected scope");
    }
    if (!current.detached) {
      current.parent.removeChild(current);
    }
  }

  private final Runnable runnable;
  private CancellationScopeImpl parent;
  private final Set<CancellationScopeImpl> children = new HashSet<>();
  /**
   * When disconnected scope has no parent and thus doesn't receive cancellation requests from it.
   */
  private boolean cancelRequested;

  private String reason;

  CancellationScopeImpl(boolean ignoreParentCancellation, Runnable runnable) {
    this(ignoreParentCancellation, runnable, current());
  }

  CancellationScopeImpl(boolean detached, Runnable runnable, CancellationScopeImpl parent) {
    this.detached = detached;
    this.runnable = runnable;
    setParent(parent);
  }

  public CancellationScopeImpl(
      boolean ignoreParentCancellation, Functions.Proc1<CancellationScope> proc) {
    this.detached = ignoreParentCancellation;
    this.runnable = () -> proc.apply(this);
    setParent(current());
  }

  private void setParent(CancellationScopeImpl parent) {
    if (parent == null) {
      detached = true;
      return;
    }
    if (!detached) {
      this.parent = parent;
      parent.addChild(this);
      if (parent.isCancelRequested()) {
        cancel(parent.getCancellationReason());
      }
    }
  }

  @Override
  public void run() {
    try {
      pushCurrent(this);
      runnable.run();
    } finally {
      popCurrent(this);
    }
  }

  @Override
  public boolean isDetached() {
    return detached;
  }

  @Override
  public void cancel() {
    cancelRequested = true;
    reason = null;
    for (CancellationScopeImpl child : children) {
      child.cancel();
    }
    if (cancellationPromise != null) {
      cancellationPromise.complete(null);
    }
  }

  @Override
  public void cancel(String reason) {
    cancelRequested = true;
    this.reason = reason;
    for (CancellationScopeImpl child : children) {
      child.cancel(reason);
    }
    if (cancellationPromise != null) {
      cancellationPromise.complete(reason);
    }
  }

  @Override
  public String getCancellationReason() {
    return reason;
  }

  @Override
  public boolean isCancelRequested() {
    return cancelRequested;
  }

  @Override
  public CompletablePromise<String> getCancellationRequest() {
    if (cancellationPromise == null) {
      cancellationPromise = Workflow.newPromise();
      if (isCancelRequested()) {
        cancellationPromise.complete(getCancellationReason());
      }
    }
    return cancellationPromise;
  }

  private void addChild(CancellationScopeImpl scope) {
    children.add(scope);
  }

  private void removeChild(CancellationScopeImpl scope) {
    if (!children.remove(scope)) {
      throw new Error("Not a child");
    }
  }
}

1	/*
2	* Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
3	*
4	* Modifications copyright (C) 2017 Uber Technologies, Inc.
5	*
6	* Licensed under the Apache License, Version 2.0 (the "License"). You may not
7	* use this file except in compliance with the License. A copy of the License is
8	* located at
9	*
10	* http://aws.amazon.com/apache2.0
11	*
12	* or in the "license" file accompanying this file. This file is distributed on
13	* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
14	* express or implied. See the License for the specific language governing
15	* permissions and limitations under the License.
16	*/
17
18	package com.uber.cadence.internal.sync;
19
20	import com.uber.cadence.workflow.CancellationScope;
21	import com.uber.cadence.workflow.CompletablePromise;
22	import com.uber.cadence.workflow.Functions;
23	import com.uber.cadence.workflow.Workflow;
24	import java.util.ArrayDeque;
25	import java.util.Deque;
26	import java.util.HashSet;
27	import java.util.Set;
28
29	class CancellationScopeImpl implements CancellationScope {
30
31	private static ThreadLocal<Deque<CancellationScopeImpl>> scopeStack =	1✔
32	ThreadLocal.withInitial(ArrayDeque::new);	1✔
33	private boolean detached;
34	private CompletablePromise<String> cancellationPromise;
35
36	static CancellationScopeImpl current() {
37	if (scopeStack.get().isEmpty()) {	1✔
38	throw new IllegalStateException("Cannot be called by non workflow thread");	×
39	}
40	return scopeStack.get().peekFirst();	1✔
41	}
42
43	private static void pushCurrent(CancellationScopeImpl scope) {
44	scopeStack.get().addFirst(scope);	1✔
45	}	1✔
46
47	private static void popCurrent(CancellationScopeImpl expected) {
48	CancellationScopeImpl current = scopeStack.get().pollFirst();	1✔
49	if (current != expected) {	1✔
50	throw new Error("Unexpected scope");	×
51	}
52	if (!current.detached) {	1✔
53	current.parent.removeChild(current);	1✔
54	}
55	}	1✔
56
57	private final Runnable runnable;
58	private CancellationScopeImpl parent;
59	private final Set<CancellationScopeImpl> children = new HashSet<>();	1✔
60	/**
61	* When disconnected scope has no parent and thus doesn't receive cancellation requests from it.
62	*/
63	private boolean cancelRequested;
64
65	private String reason;
66
67	CancellationScopeImpl(boolean ignoreParentCancellation, Runnable runnable) {
68	this(ignoreParentCancellation, runnable, current());	1✔
69	}	1✔
70
71	CancellationScopeImpl(boolean detached, Runnable runnable, CancellationScopeImpl parent) {	1✔
72	this.detached = detached;	1✔
73	this.runnable = runnable;	1✔
74	setParent(parent);	1✔
75	}	1✔
76
77	public CancellationScopeImpl(
78	boolean ignoreParentCancellation, Functions.Proc1<CancellationScope> proc) {	1✔
79	this.detached = ignoreParentCancellation;	1✔
80	this.runnable = () -> proc.apply(this);	1✔
81	setParent(current());	1✔
82	}	1✔
83
84	private void setParent(CancellationScopeImpl parent) {
85	if (parent == null) {	1✔
86	detached = true;	1✔
87	return;	1✔
88	}
89	if (!detached) {	1✔
90	this.parent = parent;	1✔
91	parent.addChild(this);	1✔
92	if (parent.isCancelRequested()) {	1✔
UNCOV 93	cancel(parent.getCancellationReason());	×
94	}
95	}
96	}	1✔
97
98	@Override
99	public void run() {
100	try {
101	pushCurrent(this);	1✔
102	runnable.run();	1✔
103	} finally {
104	popCurrent(this);	1✔
105	}
106	}	1✔
107
108	@Override
109	public boolean isDetached() {
110	return detached;	×
111	}
112
113	@Override
114	public void cancel() {
115	cancelRequested = true;	1✔
116	reason = null;	1✔
117	for (CancellationScopeImpl child : children) {	1✔
118	child.cancel();	1✔
119	}	1✔
120	if (cancellationPromise != null) {	1✔
121	cancellationPromise.complete(null);	1✔
122	}
123	}	1✔
124
125	@Override
126	public void cancel(String reason) {
127	cancelRequested = true;	1✔
128	this.reason = reason;	1✔
129	for (CancellationScopeImpl child : children) {	1✔
130	child.cancel(reason);	1✔
131	}	1✔
132	if (cancellationPromise != null) {	1✔
133	cancellationPromise.complete(reason);	1✔
134	}
135	}	1✔
136
137	@Override
138	public String getCancellationReason() {
139	return reason;	1✔
140	}
141
142	@Override
143	public boolean isCancelRequested() {
144	return cancelRequested;	1✔
145	}
146
147	@Override
148	public CompletablePromise<String> getCancellationRequest() {
149	if (cancellationPromise == null) {	1✔
150	cancellationPromise = Workflow.newPromise();	1✔
151	if (isCancelRequested()) {	1✔
152	cancellationPromise.complete(getCancellationReason());	1✔
153	}
154	}
155	return cancellationPromise;	1✔
156	}
157
158	private void addChild(CancellationScopeImpl scope) {
159	children.add(scope);	1✔
160	}	1✔
161
162	private void removeChild(CancellationScopeImpl scope) {
163	if (!children.remove(scope)) {	1✔
164	throw new Error("Not a child");	×
165	}
166	}	1✔
167	}

uber / cadence-java-client / 2428

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