Skip to content

Commit ebd2dca

Browse files
natemortnatemort
committed
Fix Error handling and add test coverage for Thrift2ProtoAdapter
1 parent b5d8ef8 commit ebd2dca

File tree

4 files changed

+1069
-164
lines changed

4 files changed

+1069
-164
lines changed

‎src/main/java/com/uber/cadence/internal/compatibility/Thrift2ProtoAdapter.java

Lines changed: 18 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -109,6 +109,7 @@
109109
import io.grpc.StatusRuntimeException;
110110
import java.util.UUID;
111111
import java.util.concurrent.CompletableFuture;
112+
import java.util.concurrent.ExecutionException;
112113
import java.util.concurrent.ForkJoinPool;
113114
import java.util.concurrent.TimeUnit;
114115
import org.apache.thrift.TException;
@@ -203,7 +204,7 @@ public RestartWorkflowExecutionResponse RestartWorkflowExecution(
203204
RestartWorkflowExecutionRequest restartRequest)
204205
throws BadRequestError, ServiceBusyError, DomainNotActiveError, LimitExceededError,
205206
EntityNotExistsError, ClientVersionNotSupportedError, TException {
206-
throw new IllegalArgumentException("unimplemented");
207+
throw new UnsupportedOperationException("unimplemented");
207208
}
208209

209210
@Override
@@ -851,7 +852,7 @@ public void DeprecateDomain(
851852
public void RestartWorkflowExecution(
852853
RestartWorkflowExecutionRequest restartRequest, AsyncMethodCallback resultHandler)
853854
throws TException {
854-
throw new IllegalArgumentException("unimplemented");
855+
throw new UnsupportedOperationException("unimplemented");
855856
}
856857

857858
@Override
@@ -880,7 +881,7 @@ public void StartWorkflowExecutionAsync(
880881
resultHandler.onComplete(
881882
ResponseMapper.startWorkflowExecutionAsyncResponse(response));
882883
} catch (Exception e) {
883-
resultHandler.onError(e);
884+
handleException(resultHandler, e);
884885
}
885886
},
886887
ForkJoinPool.commonPool());
@@ -1003,7 +1004,7 @@ public void SignalWorkflowExecution(
10031004
com.uber.cadence.api.v1.SignalWorkflowExecutionResponse response = resultFuture.get();
10041005
resultHandler.onComplete(null);
10051006
} catch (Exception e) {
1006-
resultHandler.onError(e);
1007+
handleException(resultHandler, e);
10071008
}
10081009
},
10091010
ForkJoinPool.commonPool());
@@ -1025,7 +1026,7 @@ public void SignalWithStartWorkflowExecutionAsync(
10251026
SignalWithStartWorkflowExecutionAsyncRequest signalWithStartRequest,
10261027
AsyncMethodCallback resultHandler)
10271028
throws TException {
1028-
throw new IllegalArgumentException("unimplemented");
1029+
throw new UnsupportedOperationException("unimplemented");
10291030
}
10301031

10311032
@Override
@@ -1199,7 +1200,7 @@ public void StartWorkflowExecutionWithTimeout(
11991200
com.uber.cadence.api.v1.StartWorkflowExecutionResponse response = resultFuture.get();
12001201
resultHandler.onComplete(ResponseMapper.startWorkflowExecutionResponse(response));
12011202
} catch (Exception e) {
1202-
resultHandler.onError(e);
1203+
handleException(resultHandler, e);
12031204
}
12041205
},
12051206
ForkJoinPool.commonPool());
@@ -1230,7 +1231,7 @@ public void StartWorkflowExecutionAsyncWithTimeout(
12301231
resultHandler.onComplete(
12311232
ResponseMapper.startWorkflowExecutionAsyncResponse(response));
12321233
} catch (Exception e) {
1233-
resultHandler.onError(e);
1234+
handleException(resultHandler, e);
12341235
}
12351236
},
12361237
ForkJoinPool.commonPool());
@@ -1276,7 +1277,7 @@ public void GetWorkflowExecutionHistoryWithTimeout(
12761277
resultHandler.onComplete(
12771278
ResponseMapper.getWorkflowExecutionHistoryResponse(response));
12781279
} catch (Exception e) {
1279-
resultHandler.onError(e);
1280+
handleException(resultHandler, e);
12801281
}
12811282
},
12821283
ForkJoinPool.commonPool());
@@ -1293,4 +1294,13 @@ public void SignalWorkflowExecutionWithTimeout(
12931294
throws TException {
12941295
throw new UnsupportedOperationException("not implemented");
12951296
}
1297+
1298+
private void handleException(AsyncMethodCallback callback, Exception exception) {
1299+
if (exception instanceof ExecutionException
1300+
&& exception.getCause() instanceof StatusRuntimeException) {
1301+
callback.onError(ErrorMapper.Error(((StatusRuntimeException) exception.getCause())));
1302+
} else {
1303+
callback.onError(exception);
1304+
}
1305+
}
12961306
}

‎src/main/java/com/uber/cadence/internal/compatibility/proto/serviceclient/GrpcServiceStubs.java

Lines changed: 0 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -105,7 +105,6 @@ final class GrpcServiceStubs implements IGrpcServiceStubs {
105105
shutdownChannel = true;
106106
}
107107
ClientInterceptor deadlineInterceptor = new GrpcDeadlineInterceptor(options);
108-
ClientInterceptor tracingInterceptor = newTracingInterceptor();
109108
Metadata headers = new Metadata();
110109
headers.put(LIBRARY_VERSION_HEADER_KEY, Version.LIBRARY_VERSION);
111110
headers.put(FEATURE_VERSION_HEADER_KEY, Version.FEATURE_VERSION);
@@ -124,9 +123,6 @@ final class GrpcServiceStubs implements IGrpcServiceStubs {
124123
MetadataUtils.newAttachHeadersInterceptor(headers),
125124
newOpenTelemetryInterceptor(),
126125
newOpenTracingInterceptor(options.getTracer()));
127-
if (log.isTraceEnabled()) {
128-
interceptedChannel = ClientInterceptors.intercept(interceptedChannel, tracingInterceptor);
129-
}
130126
if (options.getAuthProvider() != null) {
131127
interceptedChannel =
132128
ClientInterceptors.intercept(
@@ -318,43 +314,6 @@ private Header addTracingHeaders(Header header) {
318314
};
319315
}
320316

321-
private ClientInterceptor newTracingInterceptor() {
322-
return new ClientInterceptor() {
323-
324-
@Override
325-
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
326-
MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
327-
return new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(
328-
next.newCall(method, callOptions)) {
329-
@Override
330-
public void sendMessage(ReqT message) {
331-
log.trace("Invoking " + method.getFullMethodName() + "with input: " + message);
332-
super.sendMessage(message);
333-
}
334-
335-
@Override
336-
public void start(Listener<RespT> responseListener, Metadata headers) {
337-
Listener<RespT> listener =
338-
new ForwardingClientCallListener.SimpleForwardingClientCallListener<RespT>(
339-
responseListener) {
340-
@Override
341-
public void onMessage(RespT message) {
342-
if (method == WorkerAPIGrpc.getPollForDecisionTaskMethod()) {
343-
log.trace("Returned " + method.getFullMethodName());
344-
} else {
345-
log.trace(
346-
"Returned " + method.getFullMethodName() + " with output: " + message);
347-
}
348-
super.onMessage(message);
349-
}
350-
};
351-
super.start(listener, headers);
352-
}
353-
};
354-
}
355-
};
356-
}
357-
358317
public DomainAPIGrpc.DomainAPIBlockingStub domainBlockingStub() {
359318
return domainBlockingStub;
360319
}
@@ -481,10 +440,6 @@ public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
481440
} else if (method == WorkflowAPIGrpc.getQueryWorkflowMethod()) {
482441
duration = options.getRpcQueryTimeoutMillis();
483442
}
484-
if (log.isTraceEnabled()) {
485-
String name = method.getFullMethodName();
486-
log.trace("TimeoutInterceptor method=" + name + ", timeoutMs=" + duration);
487-
}
488443
return next.newCall(method, callOptions.withDeadlineAfter(duration, TimeUnit.MILLISECONDS));
489444
}
490445
}

‎src/test/java/com/uber/cadence/internal/compatibility/ProtoObjects.java

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1324,6 +1324,65 @@ public final class ProtoObjects {
13241324
public static final UpdateDomainResponse UPDATE_DOMAIN_RESPONSE =
13251325
UpdateDomainResponse.newBuilder().setDomain(DOMAIN).build();
13261326

1327+
public static final GetSearchAttributesRequest GET_SEARCH_ATTRIBUTES_REQUEST =
1328+
GetSearchAttributesRequest.getDefaultInstance();
1329+
1330+
public static final RegisterDomainResponse REGISTER_DOMAIN_RESPONSE =
1331+
RegisterDomainResponse.getDefaultInstance();
1332+
1333+
public static final DeprecateDomainResponse DEPRECATE_DOMAIN_RESPONSE =
1334+
DeprecateDomainResponse.getDefaultInstance();
1335+
1336+
public static final SignalWorkflowExecutionResponse SIGNAL_WORKFLOW_EXECUTION_RESPONSE =
1337+
SignalWorkflowExecutionResponse.getDefaultInstance();
1338+
1339+
public static final RequestCancelWorkflowExecutionResponse
1340+
REQUEST_CANCEL_WORKFLOW_EXECUTION_RESPONSE =
1341+
RequestCancelWorkflowExecutionResponse.getDefaultInstance();
1342+
1343+
public static final TerminateWorkflowExecutionResponse TERMINATE_WORKFLOW_EXECUTION_RESPONSE =
1344+
TerminateWorkflowExecutionResponse.getDefaultInstance();
1345+
1346+
public static final GetClusterInfoRequest GET_CLUSTER_INFO_REQUEST =
1347+
GetClusterInfoRequest.getDefaultInstance();
1348+
1349+
public static final RespondDecisionTaskFailedResponse RESPOND_DECISION_TASK_FAILED_RESPONSE =
1350+
RespondDecisionTaskFailedResponse.getDefaultInstance();
1351+
1352+
public static final RespondActivityTaskCompletedResponse
1353+
RESPOND_ACTIVITY_TASK_COMPLETED_RESPONSE =
1354+
RespondActivityTaskCompletedResponse.getDefaultInstance();
1355+
1356+
public static final RespondActivityTaskCompletedByIDResponse
1357+
RESPOND_ACTIVITY_TASK_COMPLETED_BY_ID_RESPONSE =
1358+
RespondActivityTaskCompletedByIDResponse.getDefaultInstance();
1359+
1360+
public static final RespondActivityTaskFailedResponse RESPOND_ACTIVITY_TASK_FAILED_RESPONSE =
1361+
RespondActivityTaskFailedResponse.getDefaultInstance();
1362+
1363+
public static final RespondActivityTaskFailedByIDResponse
1364+
RESPOND_ACTIVITY_TASK_FAILED_BY_ID_RESPONSE =
1365+
RespondActivityTaskFailedByIDResponse.getDefaultInstance();
1366+
1367+
public static final RespondActivityTaskCanceledResponse RESPOND_ACTIVITY_TASK_CANCELED_RESPONSE =
1368+
RespondActivityTaskCanceledResponse.getDefaultInstance();
1369+
1370+
public static final RespondActivityTaskCanceledByIDResponse
1371+
RESPOND_ACTIVITY_TASK_CANCELED_BY_ID_RESPONSE =
1372+
RespondActivityTaskCanceledByIDResponse.getDefaultInstance();
1373+
1374+
public static final RespondQueryTaskCompletedResponse RESPOND_QUERY_TASK_COMPLETED_RESPONSE =
1375+
RespondQueryTaskCompletedResponse.getDefaultInstance();
1376+
1377+
public static final ResetStickyTaskListResponse RESET_STICKY_TASK_LIST_RESPONSE =
1378+
ResetStickyTaskListResponse.getDefaultInstance();
1379+
1380+
public static final RefreshWorkflowTasksRequest REFRESH_WORKFLOW_TASKS_REQUEST =
1381+
RefreshWorkflowTasksRequest.getDefaultInstance();
1382+
1383+
public static final RefreshWorkflowTasksResponse REFRESH_WORKFLOW_TASKS_RESPONSE =
1384+
RefreshWorkflowTasksResponse.getDefaultInstance();
1385+
13271386
private ProtoObjects() {}
13281387

13291388
private static Payload payload(String value) {

0 commit comments

Comments
 (0)