Skip to content

Commit 36aab95

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

File tree

5 files changed

+1085
-124
lines changed

5 files changed

+1085
-124
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+
handleAsyncException(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+
handleAsyncException(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+
handleAsyncException(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+
handleAsyncException(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+
handleAsyncException(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 handleAsyncException(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: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
*/
1616
package com.uber.cadence.internal.compatibility.proto.serviceclient;
1717

18+
import com.google.common.annotations.VisibleForTesting;
1819
import com.google.common.base.Strings;
1920
import com.google.protobuf.ByteString;
2021
import com.uber.cadence.api.v1.*;
@@ -53,7 +54,8 @@
5354
import org.slf4j.Logger;
5455
import org.slf4j.LoggerFactory;
5556

56-
final class GrpcServiceStubs implements IGrpcServiceStubs {
57+
@VisibleForTesting
58+
public final class GrpcServiceStubs implements IGrpcServiceStubs {
5759

5860
private static final Logger log = LoggerFactory.getLogger(GrpcServiceStubs.class);
5961
private static final Metadata.Key<String> LIBRARY_VERSION_HEADER_KEY =
@@ -91,7 +93,8 @@ final class GrpcServiceStubs implements IGrpcServiceStubs {
9193
private final MetaAPIGrpc.MetaAPIBlockingStub metaBlockingStub;
9294
private final MetaAPIGrpc.MetaAPIFutureStub metaFutureStub;
9395

94-
GrpcServiceStubs(ClientOptions options) {
96+
@VisibleForTesting
97+
public GrpcServiceStubs(ClientOptions options, boolean enableLogging) {
9598
this.options = options;
9699
if (options.getGRPCChannel() != null) {
97100
this.channel = options.getGRPCChannel();
@@ -124,7 +127,7 @@ final class GrpcServiceStubs implements IGrpcServiceStubs {
124127
MetadataUtils.newAttachHeadersInterceptor(headers),
125128
newOpenTelemetryInterceptor(),
126129
newOpenTracingInterceptor(options.getTracer()));
127-
if (log.isTraceEnabled()) {
130+
if (log.isTraceEnabled() || enableLogging) {
128131
interceptedChannel = ClientInterceptors.intercept(interceptedChannel, tracingInterceptor);
129132
}
130133
if (options.getAuthProvider() != null) {

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -31,12 +31,12 @@ public interface IGrpcServiceStubs {
3131
/** Returns gRPC stubs with default options domain service. */
3232
static IGrpcServiceStubs newInstance() {
3333
return new GrpcServiceStubs(
34-
ClientOptions.newBuilder().setPort(DEFAULT_LOCAL_CADENCE_SERVER_GRPC_PORT).build());
34+
ClientOptions.newBuilder().setPort(DEFAULT_LOCAL_CADENCE_SERVER_GRPC_PORT).build(), false);
3535
}
3636

3737
/** Returns gRPC stubs with given options domain service. */
3838
static IGrpcServiceStubs newInstance(ClientOptions options) {
39-
return new GrpcServiceStubs(options);
39+
return new GrpcServiceStubs(options, false);
4040
}
4141

4242
ClientOptions getOptions();

‎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)