Skip to content

Commit 15e9801

Browse files
mfateevmfateev
authored
Fixed root workflow thread names and removed excessive error logging (#136)
* remove excessive options check * Removed excessive check * fixed gradle syntax * Fixed root thread name and removed duplicated error loggin * forgotten log statement removed
1 parent 6105d49 commit 15e9801

10 files changed

Lines changed: 52 additions & 57 deletions

File tree

‎build.gradle‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -300,8 +300,8 @@ publishing {
300300
nexusPublishing {
301301
repositories {
302302
sonatype {
303-
username = hasProperty('ossrhUsername') ? ossrhUsername : ''
304-
password = hasProperty('ossrhPassword') ? ossrhPassword : ''
303+
username = project.hasProperty('ossrhUsername') ? project.property('ossrhUsername') : ''
304+
password = project.hasProperty('ossrhPassword') ? project.property('ossrhPassword') : ''
305305
nexusUrl = uri("https://oss.sonatype.org/service/local/staging/deploy/maven2/")
306306
snapshotRepositoryUrl = uri("https://oss.sonatype.org/content/repositories/snapshots/")
307307
}

‎src/main/java/io/temporal/activity/ActivityOptions.java‎

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -176,12 +176,6 @@ public ActivityOptions build() {
176176
}
177177

178178
public ActivityOptions validateAndBuildWithDefaults() {
179-
if (scheduleToCloseTimeout == null
180-
&& (scheduleToStartTimeout == null || startToCloseTimeout == null)) {
181-
throw new IllegalStateException(
182-
"Either ScheduleToClose or both ScheduleToStart and StartToClose "
183-
+ "timeouts are required: ");
184-
}
185179
return new ActivityOptions(
186180
heartbeatTimeout,
187181
scheduleToCloseTimeout,

‎src/main/java/io/temporal/internal/sync/DeterministicRunner.java‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -55,9 +55,11 @@ static DeterministicRunner newRunner(
5555
ExecutorService threadPool,
5656
SyncDecisionContext decisionContext,
5757
Supplier<Long> clock,
58+
String rootThreadName,
5859
Runnable root,
5960
DeciderCache cache) {
60-
return new DeterministicRunnerImpl(threadPool, decisionContext, clock, root, cache);
61+
return new DeterministicRunnerImpl(
62+
threadPool, decisionContext, clock, rootThreadName, root, cache);
6163
}
6264

6365
/**
@@ -72,8 +74,10 @@ static DeterministicRunner newRunner(
7274
ExecutorService threadPool,
7375
SyncDecisionContext decisionContext,
7476
Supplier<Long> clock,
77+
String rootThreadName,
7578
Runnable root) {
76-
return new DeterministicRunnerImpl(threadPool, decisionContext, clock, root, null);
79+
return new DeterministicRunnerImpl(
80+
threadPool, decisionContext, clock, rootThreadName, root, null);
7781
}
7882

7983
/**

‎src/main/java/io/temporal/internal/sync/DeterministicRunnerImpl.java‎

Lines changed: 11 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -89,7 +89,7 @@ private NamedRunnable(String name, Runnable runnable) {
8989
}
9090

9191
private static final Logger log = LoggerFactory.getLogger(DeterministicRunnerImpl.class);
92-
static final String WORKFLOW_ROOT_THREAD_NAME = "workflow-root";
92+
static final String WORKFLOW_ROOT_THREAD_NAME = "workflow-method";
9393
private static final ThreadLocal<WorkflowThread> currentThreadThreadLocal = new ThreadLocal<>();
9494

9595
private final Lock lock = new ReentrantLock();
@@ -158,7 +158,13 @@ static void setCurrentThreadInternal(WorkflowThread coroutine) {
158158
}
159159

160160
DeterministicRunnerImpl(Supplier<Long> clock, Runnable root) {
161-
this(getDefaultThreadPool(), newDummySyncDecisionContext(), clock, root, null);
161+
this(
162+
getDefaultThreadPool(),
163+
newDummySyncDecisionContext(),
164+
clock,
165+
WORKFLOW_ROOT_THREAD_NAME,
166+
root,
167+
null);
162168
}
163169

164170
private static ThreadPoolExecutor getDefaultThreadPool() {
@@ -173,13 +179,14 @@ private static ThreadPoolExecutor getDefaultThreadPool() {
173179
SyncDecisionContext decisionContext,
174180
Supplier<Long> clock,
175181
Runnable root) {
176-
this(threadPool, decisionContext, clock, root, null);
182+
this(threadPool, decisionContext, clock, WORKFLOW_ROOT_THREAD_NAME, root, null);
177183
}
178184

179185
DeterministicRunnerImpl(
180186
ExecutorService threadPool,
181187
SyncDecisionContext decisionContext,
182188
Supplier<Long> clock,
189+
String rootName,
183190
Runnable root,
184191
DeciderCache cache) {
185192
this.threadPool = threadPool;
@@ -386,8 +393,7 @@ public void close() {
386393
throw new Error("unreachable");
387394
} catch (RuntimeException e) {
388395
log.warn(
389-
"Promise that was completedExceptionally was never accessed. "
390-
+ "The ignored exception:",
396+
"Promise completed with exception and was never accessed. The ignored exception:",
391397
CheckedExceptionWrapper.unwrap(e));
392398
}
393399
}

‎src/main/java/io/temporal/internal/sync/SyncWorkflow.java‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -112,9 +112,14 @@ public void start(HistoryEvent event, DecisionContext context) {
112112
threadPool,
113113
syncContext,
114114
context::currentTimeMillis,
115+
"interceptor-init",
115116
() -> {
116117
workflow.initialize();
117-
WorkflowInternal.newThread(false, "root", () -> workflowProc.run()).start();
118+
WorkflowInternal.newThread(
119+
false,
120+
DeterministicRunnerImpl.WORKFLOW_ROOT_THREAD_NAME,
121+
() -> workflowProc.run())
122+
.start();
118123
},
119124
cache);
120125
runner.setInterceptorHead(syncContext.getWorkflowInterceptor());

‎src/main/java/io/temporal/internal/sync/WorkflowThreadImpl.java‎

Lines changed: 0 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -108,15 +108,6 @@ public void run() {
108108
threadContext.setUnhandledException(e);
109109
}
110110
} catch (Error e) {
111-
// Error aborts decision, not fails the workflow.
112-
if (log.isErrorEnabled() && !root) {
113-
StringWriter sw = new StringWriter();
114-
PrintWriter pw = new PrintWriter(sw, true);
115-
e.printStackTrace(pw);
116-
String stackTrace = sw.getBuffer().toString();
117-
log.error(
118-
String.format("Workflow thread \"%s\" run failed with Error:\n%s", name, stackTrace));
119-
}
120111
threadContext.setUnhandledException(e);
121112
} catch (CanceledFailure e) {
122113
if (!isCancelRequested()) {
@@ -126,16 +117,6 @@ public void run() {
126117
log.debug(String.format("Workflow thread \"%s\" run cancelled", name));
127118
}
128119
} catch (Throwable e) {
129-
if (log.isWarnEnabled() && !root) {
130-
StringWriter sw = new StringWriter();
131-
PrintWriter pw = new PrintWriter(sw, true);
132-
e.printStackTrace(pw);
133-
String stackTrace = sw.getBuffer().toString();
134-
log.warn(
135-
String.format(
136-
"Workflow thread \"%s\" run failed with unhandled exception:\n%s",
137-
name, stackTrace));
138-
}
139120
threadContext.setUnhandledException(e);
140121
} finally {
141122
DeterministicRunnerImpl.setCurrentThreadInternal(null);

‎src/test/java/io/temporal/internal/sync/DeterministicRunnerTest.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -733,6 +733,7 @@ public void workflowThreadsWillEvictCacheWhenMaxThreadCountIsHit() throws Throwa
733733
new SyncDecisionContext(
734734
decisionContext, DataConverter.getDefaultInstance(), null, null),
735735
() -> 0L, // clock override
736+
"test-thread",
736737
() -> {
737738
Promise<Void> thread =
738739
Async.procedure(
@@ -759,6 +760,7 @@ public void workflowThreadsWillEvictCacheWhenMaxThreadCountIsHit() throws Throwa
759760
new SyncDecisionContext(
760761
decisionContext, DataConverter.getDefaultInstance(), null, null),
761762
() -> 0L, // clock override
763+
"test-thread",
762764
() -> {
763765
Promise<Void> thread =
764766
Async.procedure(
@@ -796,6 +798,7 @@ public void workflowThreadsWillNotEvictCacheWhenMaxThreadCountIsHit() throws Thr
796798
threadPool,
797799
null,
798800
() -> 0L, // clock override
801+
"test-thread",
799802
() -> {
800803
Promise<Void> thread =
801804
Async.procedure(
@@ -821,6 +824,7 @@ public void workflowThreadsWillNotEvictCacheWhenMaxThreadCountIsHit() throws Thr
821824
threadPool,
822825
null,
823826
() -> 0L, // clock override
827+
"test-thread",
824828
() -> {
825829
Promise<Void> thread =
826830
Async.procedure(

‎src/test/java/io/temporal/internal/sync/PromiseTest.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -154,6 +154,7 @@ public void testGetTimeout() throws Throwable {
154154
threadPool,
155155
null,
156156
() -> currentTime,
157+
"test-thread",
157158
() -> {
158159
CompletablePromise<String> f = Workflow.newPromise();
159160
trace.add("root begin");

‎src/test/java/io/temporal/worker/StickyWorkerTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -430,7 +430,7 @@ public void whenCacheIsEvictedTheWorkerCanRecover() throws Exception {
430430
// Act
431431
WorkflowClient.start(workflow::getGreeting);
432432

433-
Thread.sleep(200); // Wait for workflow to start
433+
Thread.sleep(1000); // Wait for workflow to start
434434

435435
DeciderCache cache = factory.getCache();
436436
assertNotNull(cache);

‎src/test/java/io/temporal/workflow/WorkflowTest.java‎

Lines changed: 21 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -459,7 +459,7 @@ public void testSync() {
459459
assertEquals("activity10", result);
460460
tracer.setExpected(
461461
"interceptExecuteWorkflow " + UUID_REGEXP,
462-
"newThread root",
462+
"newThread workflow-method",
463463
"newThread null",
464464
"sleep PT2S",
465465
"executeActivity ActivityWithDelay",
@@ -1499,19 +1499,19 @@ public void testContinueAsNew() {
14991499
assertEquals(111, result);
15001500
tracer.setExpected(
15011501
"interceptExecuteWorkflow " + UUID_REGEXP,
1502-
"newThread root",
1502+
"newThread workflow-method",
15031503
"continueAsNew",
15041504
"interceptExecuteWorkflow " + UUID_REGEXP,
1505-
"newThread root",
1505+
"newThread workflow-method",
15061506
"continueAsNew",
15071507
"interceptExecuteWorkflow " + UUID_REGEXP,
1508-
"newThread root",
1508+
"newThread workflow-method",
15091509
"continueAsNew",
15101510
"interceptExecuteWorkflow " + UUID_REGEXP,
1511-
"newThread root",
1511+
"newThread workflow-method",
15121512
"continueAsNew",
15131513
"interceptExecuteWorkflow " + UUID_REGEXP,
1514-
"newThread root");
1514+
"newThread workflow-method");
15151515
}
15161516

15171517
@WorkflowInterface
@@ -1546,10 +1546,10 @@ public void testContinueAsNewNoArgs() {
15461546
assertEquals("done", result);
15471547
tracer.setExpected(
15481548
"interceptExecuteWorkflow " + UUID_REGEXP,
1549-
"newThread root",
1549+
"newThread workflow-method",
15501550
"continueAsNew",
15511551
"interceptExecuteWorkflow " + UUID_REGEXP,
1552-
"newThread root");
1552+
"newThread workflow-method");
15531553
}
15541554

15551555
public static class TestAsyncActivityWorkflowImpl implements TestWorkflow1 {
@@ -2369,15 +2369,15 @@ public void testTimer() {
23692369
tracer.setExpected(
23702370
"interceptExecuteWorkflow " + UUID_REGEXP,
23712371
"registerQuery getTrace",
2372-
"newThread root",
2372+
"newThread workflow-method",
23732373
"newTimer PT0.7S",
23742374
"newTimer PT1.3S",
23752375
"newTimer PT10S");
23762376
} else {
23772377
tracer.setExpected(
23782378
"interceptExecuteWorkflow " + UUID_REGEXP,
23792379
"registerQuery getTrace",
2380-
"newThread root",
2380+
"newThread workflow-method",
23812381
"newTimer PT11M40S",
23822382
"newTimer PT21M40S",
23832383
"newTimer PT10H");
@@ -3482,10 +3482,10 @@ public void testSignalExternalWorkflow() {
34823482
tracer.setExpected(
34833483
"interceptExecuteWorkflow " + stub.getExecution().getWorkflowId(),
34843484
"registerSignal testSignal",
3485-
"newThread root",
3485+
"newThread workflow-method",
34863486
"executeChildWorkflow SignalingChild",
34873487
"interceptExecuteWorkflow " + UUID_REGEXP, // child
3488-
"newThread root",
3488+
"newThread workflow-method",
34893489
"signalExternalWorkflow " + UUID_REGEXP + " testSignal");
34903490
}
34913491

@@ -4573,7 +4573,7 @@ public void testSideEffect() {
45734573
assertEquals("activity1", result);
45744574
tracer.setExpected(
45754575
"interceptExecuteWorkflow " + UUID_REGEXP,
4576-
"newThread root",
4576+
"newThread workflow-method",
45774577
"sideEffect",
45784578
"sleep PT1S",
45794579
"executeActivity customActivity1");
@@ -4671,7 +4671,7 @@ public void testGetVersion() {
46714671
assertEquals("activity22activity1activity1activity1", result);
46724672
tracer.setExpected(
46734673
"interceptExecuteWorkflow " + UUID_REGEXP,
4674-
"newThread root",
4674+
"newThread workflow-method",
46754675
"getVersion",
46764676
"executeActivity Activity2",
46774677
"getVersion",
@@ -4984,7 +4984,7 @@ public void testGetVersionRemovedInReplay() {
49844984
assertEquals("activity22activity", result);
49854985
tracer.setExpected(
49864986
"interceptExecuteWorkflow " + UUID_REGEXP,
4987-
"newThread root",
4987+
"newThread workflow-method",
49884988
"getVersion",
49894989
"executeActivity Activity2",
49904990
"executeActivity Activity");
@@ -5022,7 +5022,7 @@ public void testGetVersionRemovedBefore() {
50225022
assertEquals("activity", result);
50235023
tracer.setExpected(
50245024
"interceptExecuteWorkflow " + UUID_REGEXP,
5025-
"newThread root",
5025+
"newThread workflow-method",
50265026
"getVersion",
50275027
"getVersion",
50285028
"getVersion",
@@ -5225,7 +5225,7 @@ public void testUUIDAndRandom() {
52255225
assertEquals("foo10", result);
52265226
tracer.setExpected(
52275227
"interceptExecuteWorkflow " + UUID_REGEXP,
5228-
"newThread root",
5228+
"newThread workflow-method",
52295229
"sideEffect",
52305230
"sideEffect",
52315231
"executeActivity Activity2");
@@ -5934,15 +5934,15 @@ public void testSaga() {
59345934
sagaWorkflow.execute(taskList, false);
59355935
tracer.setExpected(
59365936
"interceptExecuteWorkflow " + UUID_REGEXP,
5937-
"newThread root",
5937+
"newThread workflow-method",
59385938
"executeActivity customActivity1",
59395939
"executeChildWorkflow TestMultiargsWorkflowsFunc",
59405940
"interceptExecuteWorkflow " + UUID_REGEXP,
5941-
"newThread root",
5941+
"newThread workflow-method",
59425942
"executeActivity ThrowIO",
59435943
"executeChildWorkflow TestCompensationWorkflow",
59445944
"interceptExecuteWorkflow " + UUID_REGEXP,
5945-
"newThread root",
5945+
"newThread workflow-method",
59465946
"executeActivity Activity2");
59475947
}
59485948

@@ -6063,7 +6063,7 @@ public void testUpsertSearchAttributes() {
60636063
assertEquals("done", result);
60646064
tracer.setExpected(
60656065
"interceptExecuteWorkflow " + UUID_REGEXP,
6066-
"newThread root",
6066+
"newThread workflow-method",
60676067
"upsertSearchAttributes",
60686068
"executeActivity Activity");
60696069
}

0 commit comments

Comments
 (0)