Some PR fixes

googleapis · gcf-merge-on-green · Mar 27, 2024 · Mar 6, 2024 · Mar 6, 2024 · Mar 6, 2024
commit 08bcd36e3e3a0fb87447ce110e85dc80767ff7c9
@@ -104,6 +104,7 @@
 import com.google.cloud.bigtable.data.v2.stub.mutaterows.BulkMutateRowsUserFacingCallable;
 import com.google.cloud.bigtable.data.v2.stub.mutaterows.MutateRowsAttemptResult;
 import com.google.cloud.bigtable.data.v2.stub.mutaterows.MutateRowsBatchingDescriptor;
+import com.google.cloud.bigtable.data.v2.stub.mutaterows.MutateRowsPartialErrorRetryAlgorithm;
 import com.google.cloud.bigtable.data.v2.stub.mutaterows.MutateRowsRetryingCallable;
 import com.google.cloud.bigtable.data.v2.stub.readrows.FilterMarkerRowsCallable;
 import com.google.cloud.bigtable.data.v2.stub.readrows.ReadRowsBatchingDescriptor;
@@ -113,7 +114,6 @@
 import com.google.cloud.bigtable.data.v2.stub.readrows.ReadRowsUserCallable;
 import com.google.cloud.bigtable.data.v2.stub.readrows.RowMergingCallable;
 import com.google.cloud.bigtable.gaxx.retrying.ApiResultRetryAlgorithm;
-import com.google.cloud.bigtable.gaxx.retrying.MutateRowsPartialErrorRetryAlgorithm;
 import com.google.cloud.bigtable.gaxx.retrying.RetryInfoRetryAlgorithm;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.MoreObjects;
@@ -679,12 +679,13 @@ public Map<String, String> extract(MutateRowRequest mutateRowRequest) {
  * been applied, are filtered from the next attempt. Also, any entries that failed with a
  * nontransient error, are filtered from the next attempt. This will continue until there
  * are no more entries or there are no more retry attempts left.
- * <li>Wrap batch failures in a {@link
- * com.google.cloud.bigtable.data.v2.models.MutateRowsException}.
+ * <li>Wrap batch failures in a {@link MutateRowsAttemptResult}.
  * <li>Add tracing & metrics.
  * </ul>
  *
  * This callable returns an internal type {@link MutateRowsAttemptResult}.
+ *
+ * <p>This function should not be exposed to external users, as it could cause a data loss.
  */
  private UnaryCallable<BulkMutation, MutateRowsAttemptResult> createMutateRowsBaseCallable() {
  ServerStreamingCallable<MutateRowsRequest, MutateRowsResponse> base =
@@ -1175,6 +1176,11 @@ public UnaryCallable<BulkMutation, Void> bulkMutateRowsCallable() {
  return externalBulkMutateRowsCallable;
  }
 
+ @InternalApi
+ public UnaryCallable<BulkMutation, MutateRowsAttemptResult> internalBulkMutateRowsCallable() {
+ return bulkMutateRowsCallable;
+ }
+
  /**
  * Returns the callable chain created in {@link #createCheckAndMutateRowCallable()} during stub
  * construction.

@@ -13,13 +13,12 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package com.google.cloud.bigtable.gaxx.retrying;
+package com.google.cloud.bigtable.data.v2.stub.mutaterows;
 
 import com.google.api.core.InternalApi;
 import com.google.api.gax.retrying.ResultRetryAlgorithmWithContext;
 import com.google.api.gax.retrying.RetryingContext;
 import com.google.api.gax.retrying.TimedAttemptSettings;
-import com.google.cloud.bigtable.data.v2.stub.mutaterows.MutateRowsAttemptResult;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 /**
@@ -39,12 +38,11 @@ public MutateRowsPartialErrorRetryAlgorithm(
  @Override
  public boolean shouldRetry(
  Throwable previousThrowable, MutateRowsAttemptResult previousResponse) {
- if (previousResponse == null) {
- return retryAlgorithm.shouldRetry(previousThrowable, null);
- }
- if (!previousResponse.failedMutations.isEmpty()) {
+ // handle partial retryable failures
+ if (previousResponse != null && !previousResponse.failedMutations.isEmpty()) {
  return previousResponse.isRetryable;
  }
+ // business as usual
  return retryAlgorithm.shouldRetry(previousThrowable, previousResponse);
  }
 
@@ -53,12 +51,11 @@ public boolean shouldRetry(
  @Nullable RetryingContext context,
  Throwable previousThrowable,
  MutateRowsAttemptResult previousResponse) {
- if (previousResponse == null) {
- return retryAlgorithm.shouldRetry(context, previousThrowable, null);
- }
- if (!previousResponse.failedMutations.isEmpty()) {
+ // handle partial retryable failures
+ if (previousResponse != null && !previousResponse.failedMutations.isEmpty()) {
  return previousResponse.isRetryable;
  }
+ // business as usual
  return retryAlgorithm.shouldRetry(context, previousThrowable, previousResponse);
  }
 

@@ -453,8 +453,7 @@ public Object answer(InvocationOnMock invocation) {
  try (Batcher<RowMutationEntry, Void> batcher =
  new BatcherImpl<>(
  batchingDescriptor,
- new ReverseErrorsConverter(
- stub.bulkMutateRowsCallable().withDefaultCallContext(defaultContext)),
+ stub.internalBulkMutateRowsCallable().withDefaultCallContext(defaultContext),
  BulkMutation.create(TABLE_ID),
  settings.getStubSettings().bulkMutateRowsSettings().getBatchingSettings(),
  Executors.newSingleThreadScheduledExecutor(),
@@ -480,22 +479,4 @@ public Object answer(InvocationOnMock invocation) {
  private static <T> StreamObserver<T> anyObserver(Class<T> returnType) {
  return (StreamObserver<T>) any(returnType);
  }
-
- private class ReverseErrorsConverter
- extends UnaryCallable<BulkMutation, MutateRowsAttemptResult> {
-
- private final UnaryCallable<BulkMutation, Void> innerCallable;
-
- ReverseErrorsConverter(UnaryCallable<BulkMutation, Void> callable) {
- this.innerCallable = callable;
- }
-
- @Override
- public ApiFuture<MutateRowsAttemptResult> futureCall(
- BulkMutation request, ApiCallContext context) {
- ApiFuture<Void> future = innerCallable.futureCall(request, context);
- return ApiFutures.transform(
- future, result -> new MutateRowsAttemptResult(), MoreExecutors.directExecutor());
- }
- }
 }
@@ -131,7 +131,14 @@ public void splitResponsePartialErrorsTest() throws Exception {
  assertThat(batchResponse.get(0).getResultFuture().isDone()).isTrue();
  assertThat(batchResponse.get(1).getResultFuture().isDone()).isTrue();
 
- batchResponse.get(1).getResultFuture().get();
+ Throwable unexpectedError = null;
+ try {
+ batchResponse.get(1).getResultFuture().get();
+
+ } catch (Throwable t) {
+ unexpectedError = t;
+ }
+ assertThat(unexpectedError).isNull();
 
  Throwable actualError = null;
  try {