Skip to content

Commit 78a78ff

Browse files
halakarakihalakaraki
authored
Merge branch 'master' into add-metrics-corrupted-signals
2 parents 75ba76b + 56f26ef commit 78a78ff

File tree

8 files changed

+89
-63
lines changed

8 files changed

+89
-63
lines changed

‎src/main/java/com/uber/cadence/converter/DataConverterException.java‎

Lines changed: 23 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -28,42 +28,40 @@
2828
@SuppressWarnings("serial")
2929
public class DataConverterException extends RuntimeException {
3030

31-
private Type[] valueTypes;
32-
33-
private String content;
34-
3531
public DataConverterException(byte[] content, Type[] valueTypes, Throwable cause) {
36-
super(cause);
37-
this.valueTypes = valueTypes;
38-
setContent(content);
32+
super(toMessage(null, content, valueTypes), cause);
3933
}
4034

41-
public DataConverterException(byte[] content, Throwable cause) {
35+
public DataConverterException(Exception cause) {
4236
super(cause);
43-
setContent(content);
4437
}
4538

46-
public DataConverterException(Exception e) {
47-
super(e);
39+
public DataConverterException(String message, byte[] content, Type[] valueTypes) {
40+
super(toMessage(message, content, valueTypes));
4841
}
4942

50-
private void setContent(byte[] content) {
51-
if (content != null) {
52-
// Limit size of the string.
53-
int maxIndex = Math.min(content.length, 255);
54-
this.content = new String(content, 0, maxIndex, StandardCharsets.UTF_8);
43+
private static String toMessage(String message, byte[] content, Type[] valueTypes) {
44+
if (content == null && valueTypes == null) {
45+
return message;
46+
}
47+
StringBuilder result = new StringBuilder();
48+
if (message != null && message.length() > 0) {
49+
result.append(message);
50+
result.append(" ");
5551
}
52+
result.append("when parsing:\"");
53+
result.append(truncateContent(content));
54+
result.append("\" into following types: ");
55+
result.append(Arrays.toString(valueTypes));
56+
return result.toString();
5657
}
5758

58-
@Override
59-
public String getMessage() {
60-
if (content == null && valueTypes == null) {
61-
return super.getMessage();
59+
private static String truncateContent(byte[] content) {
60+
if (content == null) {
61+
return "";
6262
}
63-
return super.getMessage()
64-
+ " when parsing:\""
65-
+ content
66-
+ "\" into following types: "
67-
+ Arrays.toString(valueTypes);
63+
// Limit size of the string.
64+
int maxIndex = Math.min(content.length, 255);
65+
return new String(content, 0, maxIndex, StandardCharsets.UTF_8);
6866
}
6967
}

‎src/main/java/com/uber/cadence/converter/JsonDataConverter.java‎

Lines changed: 17 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -110,7 +110,7 @@ public JsonDataConverter(Function<GsonBuilder, GsonBuilder> builderInterceptor)
110110
@Override
111111
public byte[] toData(Object... values) throws DataConverterException {
112112
if (values == null || values.length == 0) {
113-
return EMPTY_BLOB;
113+
return null;
114114
}
115115
try {
116116
if (values.length == 1) {
@@ -144,29 +144,34 @@ public <T> T fromData(byte[] content, Class<T> valueClass, Type valueType)
144144
}
145145
return gson.fromJson(new String(content, StandardCharsets.UTF_8), valueType);
146146
} catch (Exception e) {
147-
throw new DataConverterException(content, e);
147+
throw new DataConverterException(content, new Type[] {valueType}, e);
148148
}
149149
}
150150

151151
@Override
152-
public Object[] fromDataArray(byte[] content, Type... valueType) throws DataConverterException {
152+
public Object[] fromDataArray(byte[] content, Type... valueTypes) throws DataConverterException {
153153
try {
154-
if ((content == null || content.length == 0)
155-
&& (valueType == null || valueType.length == 0)) {
156-
return EMPTY_OBJECT_ARRAY;
154+
if (content == null) {
155+
if (valueTypes.length == 0) {
156+
return EMPTY_OBJECT_ARRAY;
157+
}
158+
throw new DataConverterException(
159+
"Content doesn't match expected arguments", content, valueTypes);
157160
}
158-
if (valueType.length == 1) {
159-
Object result = gson.fromJson(new String(content, StandardCharsets.UTF_8), valueType[0]);
161+
if (valueTypes.length == 1) {
162+
Object result = gson.fromJson(new String(content, StandardCharsets.UTF_8), valueTypes[0]);
160163
return new Object[] {result};
161164
}
162165
JsonArray array = parser.parse(new String(content, StandardCharsets.UTF_8)).getAsJsonArray();
163-
Object[] result = new Object[valueType.length];
164-
for (int i = 0; i < valueType.length; i++) {
165-
result[i] = gson.fromJson(array.get(i), valueType[i]);
166+
Object[] result = new Object[valueTypes.length];
167+
for (int i = 0; i < valueTypes.length; i++) {
168+
result[i] = gson.fromJson(array.get(i), valueTypes[i]);
166169
}
167170
return result;
171+
} catch (DataConverterException e) {
172+
throw e;
168173
} catch (Exception e) {
169-
throw new DataConverterException(content, valueType, e);
174+
throw new DataConverterException(content, valueTypes, e);
170175
}
171176
}
172177

‎src/main/java/com/uber/cadence/internal/external/GenericWorkflowClientExternalImpl.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -86,8 +86,9 @@ private WorkflowExecution startWorkflowInternal(StartWorkflowExecutionParameters
8686
throws WorkflowExecutionAlreadyStartedError {
8787
StartWorkflowExecutionRequest request = new StartWorkflowExecutionRequest();
8888
request.setDomain(domain);
89-
90-
request.setInput(startParameters.getInput());
89+
if (startParameters.getInput() != null) {
90+
request.setInput(startParameters.getInput());
91+
}
9192
request.setExecutionStartToCloseTimeoutSeconds(
9293
(int) startParameters.getExecutionStartToCloseTimeoutSeconds());
9394
request.setTaskStartToCloseTimeoutSeconds(

‎src/main/java/com/uber/cadence/internal/testservice/StateMachine.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717

1818
package com.uber.cadence.internal.testservice;
1919

20+
import com.uber.cadence.BadRequestError;
2021
import com.uber.cadence.InternalServiceError;
2122
import com.uber.cadence.internal.testservice.StateMachines.Action;
2223
import com.uber.cadence.internal.testservice.StateMachines.State;
@@ -41,7 +42,8 @@ final class StateMachine<Data> {
4142
@FunctionalInterface
4243
interface Callback<D, R> {
4344

44-
void apply(RequestContext ctx, D data, R request, long referenceId) throws InternalServiceError;
45+
void apply(RequestContext ctx, D data, R request, long referenceId)
46+
throws InternalServiceError, BadRequestError;
4547
}
4648

4749
private static class Transition {
@@ -151,7 +153,7 @@ <V> StateMachine<Data> add(State from, Action action, State to, Callback<Data, V
151153
}
152154

153155
<V> void action(Action action, RequestContext context, V request, long referenceId)
154-
throws InternalServiceError {
156+
throws InternalServiceError, BadRequestError {
155157
Transition transition = new Transition(state, action);
156158
TransitionDestination<Data> destination = transitions.get(transition);
157159
if (destination == null) {

‎src/main/java/com/uber/cadence/internal/testservice/StateMachines.java‎

Lines changed: 29 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@
4343
import com.uber.cadence.ActivityTaskScheduledEventAttributes;
4444
import com.uber.cadence.ActivityTaskStartedEventAttributes;
4545
import com.uber.cadence.ActivityTaskTimedOutEventAttributes;
46+
import com.uber.cadence.BadRequestError;
4647
import com.uber.cadence.CancelTimerDecisionAttributes;
4748
import com.uber.cadence.CancelWorkflowExecutionDecisionAttributes;
4849
import com.uber.cadence.ChildWorkflowExecutionCanceledEventAttributes;
@@ -417,7 +418,6 @@ private static void initiateChildWorkflow(
417418
data.initiatedEvent = a;
418419
StartWorkflowExecutionRequest startChild =
419420
new StartWorkflowExecutionRequest()
420-
.setInput(d.getInput())
421421
.setDomain(d.getDomain() == null ? ctx.getDomain() : d.getDomain())
422422
.setExecutionStartToCloseTimeoutSeconds(
423423
d.getExecutionStartToCloseTimeoutSeconds())
@@ -426,6 +426,9 @@ private static void initiateChildWorkflow(
426426
.setWorkflowId(d.getWorkflowId())
427427
.setWorkflowIdReusePolicy(d.getWorkflowIdReusePolicy())
428428
.setWorkflowType(d.getWorkflowType());
429+
if (d.isSetInput()) {
430+
startChild.setInput(d.getInput());
431+
}
429432
addStartChildTask(ctx, data, initiatedEventId, startChild);
430433
});
431434
}
@@ -461,16 +464,31 @@ private static void addStartChildTask(
461464
}
462465

463466
private static void startWorkflow(
464-
RequestContext ctx, WorkflowData data, StartWorkflowExecutionRequest request, long notUsed) {
465-
WorkflowExecutionStartedEventAttributes a =
466-
new WorkflowExecutionStartedEventAttributes()
467-
.setIdentity(request.getIdentity())
468-
.setTaskStartToCloseTimeoutSeconds(request.getTaskStartToCloseTimeoutSeconds())
469-
.setWorkflowType(request.getWorkflowType())
470-
.setTaskList(request.getTaskList())
471-
.setExecutionStartToCloseTimeoutSeconds(
472-
request.getExecutionStartToCloseTimeoutSeconds())
473-
.setInput(request.getInput());
467+
RequestContext ctx, WorkflowData data, StartWorkflowExecutionRequest request, long notUsed)
468+
throws BadRequestError {
469+
WorkflowExecutionStartedEventAttributes a = new WorkflowExecutionStartedEventAttributes();
470+
if (request.isSetIdentity()) {
471+
a.setIdentity(request.getIdentity());
472+
}
473+
if (!request.isSetTaskStartToCloseTimeoutSeconds()) {
474+
throw new BadRequestError("missing taskStartToCloseTimeoutSeconds");
475+
}
476+
a.setTaskStartToCloseTimeoutSeconds(request.getTaskStartToCloseTimeoutSeconds());
477+
if (!request.isSetWorkflowType()) {
478+
throw new BadRequestError("missing workflowType");
479+
}
480+
a.setWorkflowType(request.getWorkflowType());
481+
if (!request.isSetTaskList()) {
482+
throw new BadRequestError("missing taskList");
483+
}
484+
a.setTaskList(request.getTaskList());
485+
if (!request.isSetExecutionStartToCloseTimeoutSeconds()) {
486+
throw new BadRequestError("missing executionStartToCloseTimeoutSeconds");
487+
}
488+
a.setExecutionStartToCloseTimeoutSeconds(request.getExecutionStartToCloseTimeoutSeconds());
489+
if (request.isSetInput()) {
490+
a.setInput(request.getInput());
491+
}
474492
HistoryEvent event =
475493
new HistoryEvent()
476494
.setEventType(EventType.WorkflowExecutionStarted)

‎src/main/java/com/uber/cadence/internal/testservice/TestWorkflowMutableStateImpl.java‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -386,7 +386,7 @@ private void processRecordMarker(
386386

387387
private void processCancelTimer(
388388
RequestContext ctx, CancelTimerDecisionAttributes d, long decisionTaskCompletedId)
389-
throws InternalServiceError {
389+
throws InternalServiceError, BadRequestError {
390390
String timerId = d.getTimerId();
391391
StateMachine<TimerData> timer = timers.get(timerId);
392392
if (timer == null) {
@@ -410,7 +410,7 @@ private void processRequestCancelActivityTask(
410410
RequestContext ctx,
411411
RequestCancelActivityTaskDecisionAttributes a,
412412
long decisionTaskCompletedId)
413-
throws InternalServiceError {
413+
throws InternalServiceError, BadRequestError {
414414
String activityId = a.getActivityId();
415415
StateMachine<?> activity = activities.get(activityId);
416416
if (activity == null) {
@@ -508,7 +508,7 @@ private void processSignalExternalWorkflowExecution(
508508
RequestContext ctx,
509509
SignalExternalWorkflowExecutionDecisionAttributes a,
510510
long decisionTaskCompletedId)
511-
throws InternalServiceError {
511+
throws InternalServiceError, BadRequestError {
512512
String signalId = UUID.randomUUID().toString();
513513
StateMachine<SignalExternalData> signalStateMachine =
514514
StateMachines.newSignalExternalStateMachine();
@@ -718,7 +718,7 @@ private void fireTimer(String timerId) {
718718

719719
private void processFailWorkflowExecution(
720720
RequestContext ctx, FailWorkflowExecutionDecisionAttributes d, long decisionTaskCompletedId)
721-
throws InternalServiceError {
721+
throws InternalServiceError, BadRequestError {
722722
workflow.action(StateMachines.Action.FAIL, ctx, d, decisionTaskCompletedId);
723723
if (parent.isPresent()) {
724724
ctx.lockTimer(); // unlocked by the parent
@@ -750,7 +750,7 @@ private void processCompleteWorkflowExecution(
750750
RequestContext ctx,
751751
CompleteWorkflowExecutionDecisionAttributes d,
752752
long decisionTaskCompletedId)
753-
throws InternalServiceError {
753+
throws InternalServiceError, BadRequestError {
754754
workflow.action(StateMachines.Action.COMPLETE, ctx, d, decisionTaskCompletedId);
755755
if (parent.isPresent()) {
756756
ctx.lockTimer(); // unlocked by the parent
@@ -780,7 +780,7 @@ private void processCompleteWorkflowExecution(
780780

781781
private void processCancelWorkflowExecution(
782782
RequestContext ctx, CancelWorkflowExecutionDecisionAttributes d, long decisionTaskCompletedId)
783-
throws InternalServiceError {
783+
throws InternalServiceError, BadRequestError {
784784
workflow.action(StateMachines.Action.CANCEL, ctx, d, decisionTaskCompletedId);
785785
if (parent.isPresent()) {
786786
ctx.lockTimer(); // unlocked by the parent
@@ -863,7 +863,7 @@ public void startWorkflow() throws InternalServiceError, BadRequestError {
863863
}
864864
}
865865

866-
private void scheduleDecision(RequestContext ctx) throws InternalServiceError {
866+
private void scheduleDecision(RequestContext ctx) throws InternalServiceError, BadRequestError {
867867
if (decision != null) {
868868
if (decision.getState() == StateMachines.State.INITIATED) {
869869
return; // No need to schedule again

‎src/main/java/com/uber/cadence/internal/testservice/TestWorkflowService.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -450,7 +450,6 @@ public String continueAsNew(
450450
throws InternalServiceError, BadRequestError {
451451
StartWorkflowExecutionRequest startRequest =
452452
new StartWorkflowExecutionRequest()
453-
.setInput(a.getInput())
454453
.setWorkflowType(a.getWorkflowType())
455454
.setExecutionStartToCloseTimeoutSeconds(a.getExecutionStartToCloseTimeoutSeconds())
456455
.setTaskStartToCloseTimeoutSeconds(a.getTaskStartToCloseTimeoutSeconds())
@@ -459,6 +458,9 @@ public String continueAsNew(
459458
.setWorkflowId(executionId.getWorkflowId().getWorkflowId())
460459
.setWorkflowIdReusePolicy(previousRunStartRequest.getWorkflowIdReusePolicy())
461460
.setIdentity(identity);
461+
if (a.isSetInput()) {
462+
startRequest.setInput(a.getInput());
463+
}
462464
lock.lock();
463465
try {
464466
StartWorkflowExecutionResponse response =

‎src/test/java/com/uber/cadence/workflow/WorkflowTest.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -633,7 +633,7 @@ public void testSyncUntypedAndStackTrace() throws InterruptedException {
633633
WorkflowStub workflowStub =
634634
workflowClient.newUntypedWorkflowStub(
635635
"TestWorkflow1::execute", newWorkflowOptionsBuilder(taskList).build());
636-
WorkflowExecution execution = workflowStub.start();
636+
WorkflowExecution execution = workflowStub.start(taskList);
637637
Thread.sleep(500);
638638
String stackTrace = workflowStub.query(WorkflowClient.QUERY_TYPE_STACK_TRCE, String.class);
639639
assertTrue(stackTrace, stackTrace.contains("WorkflowTest$TestSyncWorkflowImpl.execute"));
@@ -653,7 +653,7 @@ public void testWorkflowCancellation() {
653653
WorkflowStub client =
654654
workflowClient.newUntypedWorkflowStub(
655655
"TestWorkflow1::execute", newWorkflowOptionsBuilder(taskList).build());
656-
client.start();
656+
client.start(taskList);
657657
client.cancel();
658658
try {
659659
client.getResult(String.class);
@@ -724,7 +724,7 @@ public void testDetachedScope() throws InterruptedException {
724724
WorkflowStub client =
725725
workflowClient.newUntypedWorkflowStub(
726726
"TestWorkflow1::execute", newWorkflowOptionsBuilder(taskList).build());
727-
client.start();
727+
client.start(taskList);
728728
Thread.sleep(500); // To let activityWithDelay start.
729729
client.cancel();
730730
try {

0 commit comments

Comments
 (0)