2409

Committed 03 Jul 2024 08:33PM UTC coverage: 61.467% (-0.05%) from 61.518%

Build # 2409

Build Type

push

buildkite

Committed by

web-flow

Commit Message

Avoid consuming ByteBuffers (#913)

A ByteBuffer is a pointer to a byte[] with a starting position, a current position, and a limit. Any function that reads from its contents updates the current position. Both TracingPropagator and WorkflowUtils copy the entirety of its contents, and in doing so they mutate the current position. WorkflowUtils resets it afterwards but this still isn't thread-safe as another thread may be trying to read it.

By duplicating the ByteBuffer (copying only the metadata, not the actual contents) we avoid modifying it. It doesn't seem likely that there's real impact in either of these cases beyond unit tests, where these ByteBuffers stick around in the workflow history and are repeatedly serialized/deserialized. Modifying them during serialization can create test flakiness as that can trigger exceptions.

Run Details

2 of 2 new or added lines in 2 files covered. (100.0%)

10 existing lines in 4 files now uncovered.

11972 of 19477 relevant lines covered (61.47%)

0.61 hits per line

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

0.0

/src/main/java/com/uber/cadence/internal/compatibility/thrift/ErrorMapper.java

/*
 *  Modifications Copyright (c) 2017-2021 Uber Technologies Inc.
 *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
 *
 *  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.compatibility.thrift;

import com.uber.cadence.AccessDeniedError;
import com.uber.cadence.CancellationAlreadyRequestedError;
import com.uber.cadence.ClientVersionNotSupportedError;
import com.uber.cadence.DomainAlreadyExistsError;
import com.uber.cadence.DomainNotActiveError;
import com.uber.cadence.EntityNotExistsError;
import com.uber.cadence.FeatureNotEnabledError;
import com.uber.cadence.InternalDataInconsistencyError;
import com.uber.cadence.InternalServiceError;
import com.uber.cadence.LimitExceededError;
import com.uber.cadence.ServiceBusyError;
import com.uber.cadence.WorkflowExecutionAlreadyCompletedError;
import com.uber.cadence.WorkflowExecutionAlreadyStartedError;
import io.grpc.Metadata;
import io.grpc.StatusRuntimeException;
import org.apache.thrift.TException;

public class ErrorMapper {

  public static TException Error(StatusRuntimeException ex) {
    String details = getErrorDetails(ex);
    switch (ex.getStatus().getCode()) {
      case PERMISSION_DENIED:
        return new AccessDeniedError(ex.getMessage());
      case INTERNAL:
        return new InternalServiceError(ex.getMessage());
      case NOT_FOUND:
        {
          if ("EntityNotExistsError".equals(details)
              && ex.getMessage().contains("already completed.")) {
            return new WorkflowExecutionAlreadyCompletedError(ex.getMessage());
          } else {
            // TODO add cluster info
            return new EntityNotExistsError(ex.getMessage());
          }
        }
      case ALREADY_EXISTS:
        {
          switch (details) {
            case "CancellationAlreadyRequestedError":
              return new CancellationAlreadyRequestedError(ex.getMessage());
            case "DomainAlreadyExistsError":
              return new DomainAlreadyExistsError(ex.getMessage());
            case "WorkflowExecutionAlreadyStartedError":
              {
                // TODO add started wf info
                WorkflowExecutionAlreadyStartedError e = new WorkflowExecutionAlreadyStartedError();
                e.setMessage(ex.getMessage());
                return e;
              }
          }
        }
      case DATA_LOSS:
        return new InternalDataInconsistencyError(ex.getMessage());
      case FAILED_PRECONDITION:
        switch (details) {
            // TODO add infos
          case "ClientVersionNotSupportedError":
            return new ClientVersionNotSupportedError();
          case "FeatureNotEnabledError":
            return new FeatureNotEnabledError();
          case "DomainNotActiveError":
            {
              DomainNotActiveError e = new DomainNotActiveError();
              e.setMessage(ex.getMessage());
              return e;
            }
        }
      case RESOURCE_EXHAUSTED:
        switch (details) {
          case "LimitExceededError":
            return new LimitExceededError(ex.getMessage());
          case "ServiceBusyError":
            return new ServiceBusyError(ex.getMessage());
        }
      case UNKNOWN:
        return new TException(ex);
      default:
        // If error does not match anything, return raw grpc status error
        // There are some code that casts error to grpc status to check for deadline exceeded status
        return new TException(ex);
    }
  }

  static String getErrorDetails(StatusRuntimeException ex) {
    {
      Metadata trailer = ex.getTrailers();
      Metadata.Key<String> key =
          Metadata.Key.of("rpc-application-error-name", Metadata.ASCII_STRING_MARSHALLER);
      if (trailer != null && trailer.containsKey(key)) {
        return trailer.get(key);
      } else {
        return "";
      }
    }
  }
}

1	/*
2	* Modifications Copyright (c) 2017-2021 Uber Technologies Inc.
3	* Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
4	*
5	* Licensed under the Apache License, Version 2.0 (the "License"). You may not
6	* use this file except in compliance with the License. A copy of the License is
7	* located at
8	*
9	* http://aws.amazon.com/apache2.0
10	*
11	* or in the "license" file accompanying this file. This file is distributed on
12	* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
13	* express or implied. See the License for the specific language governing
14	* permissions and limitations under the License.
15	*/
16	package com.uber.cadence.internal.compatibility.thrift;
17
18	import com.uber.cadence.AccessDeniedError;
19	import com.uber.cadence.CancellationAlreadyRequestedError;
20	import com.uber.cadence.ClientVersionNotSupportedError;
21	import com.uber.cadence.DomainAlreadyExistsError;
22	import com.uber.cadence.DomainNotActiveError;
23	import com.uber.cadence.EntityNotExistsError;
24	import com.uber.cadence.FeatureNotEnabledError;
25	import com.uber.cadence.InternalDataInconsistencyError;
26	import com.uber.cadence.InternalServiceError;
27	import com.uber.cadence.LimitExceededError;
28	import com.uber.cadence.ServiceBusyError;
29	import com.uber.cadence.WorkflowExecutionAlreadyCompletedError;
30	import com.uber.cadence.WorkflowExecutionAlreadyStartedError;
31	import io.grpc.Metadata;
32	import io.grpc.StatusRuntimeException;
33	import org.apache.thrift.TException;
34
35	public class ErrorMapper {	×
36
37	public static TException Error(StatusRuntimeException ex) {
38	String details = getErrorDetails(ex);	×
39	switch (ex.getStatus().getCode()) {	×
40	case PERMISSION_DENIED:
41	return new AccessDeniedError(ex.getMessage());	×
42	case INTERNAL:
43	return new InternalServiceError(ex.getMessage());	×
44	case NOT_FOUND:
45	{
46	if ("EntityNotExistsError".equals(details)	×
47	&& ex.getMessage().contains("already completed.")) {	×
48	return new WorkflowExecutionAlreadyCompletedError(ex.getMessage());	×
49	} else {
50	// TODO add cluster info
51	return new EntityNotExistsError(ex.getMessage());	×
52	}
53	}
54	case ALREADY_EXISTS:
55	{
56	switch (details) {	×
57	case "CancellationAlreadyRequestedError":
58	return new CancellationAlreadyRequestedError(ex.getMessage());	×
59	case "DomainAlreadyExistsError":
60	return new DomainAlreadyExistsError(ex.getMessage());	×
61	case "WorkflowExecutionAlreadyStartedError":
62	{
63	// TODO add started wf info
64	WorkflowExecutionAlreadyStartedError e = new WorkflowExecutionAlreadyStartedError();	×
65	e.setMessage(ex.getMessage());	×
66	return e;	×
67	}
68	}
69	}
70	case DATA_LOSS:
71	return new InternalDataInconsistencyError(ex.getMessage());	×
72	case FAILED_PRECONDITION:
73	switch (details) {	×
74	// TODO add infos
75	case "ClientVersionNotSupportedError":
76	return new ClientVersionNotSupportedError();	×
77	case "FeatureNotEnabledError":
78	return new FeatureNotEnabledError();	×
79	case "DomainNotActiveError":
80	{
81	DomainNotActiveError e = new DomainNotActiveError();	×
82	e.setMessage(ex.getMessage());	×
83	return e;	×
84	}
85	}
86	case RESOURCE_EXHAUSTED:
87	switch (details) {	×
88	case "LimitExceededError":
89	return new LimitExceededError(ex.getMessage());	×
90	case "ServiceBusyError":
91	return new ServiceBusyError(ex.getMessage());	×
92	}
93	case UNKNOWN:
94	return new TException(ex);	×
95	default:
96	// If error does not match anything, return raw grpc status error
97	// There are some code that casts error to grpc status to check for deadline exceeded status
98	return new TException(ex);	×
99	}
100	}
101
102	static String getErrorDetails(StatusRuntimeException ex) {
103	{
104	Metadata trailer = ex.getTrailers();	×
105	Metadata.Key<String> key =	×
106	Metadata.Key.of("rpc-application-error-name", Metadata.ASCII_STRING_MARSHALLER);	×
107	if (trailer != null && trailer.containsKey(key)) {	×
108	return trailer.get(key);	×
109	} else {
110	return "";	×
111	}
112	}
113	}
114	}

uber / cadence-java-client / 2409

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