BulkMutation failure: MutateRowsException: Some mutations failed to apply

[santy6781]

I am working on bigtable java api

<groupId>com.google.cloud.bigtable</groupId>

<artifactId>bigtable-client-core</artifactId>

<version>1.12.0</version>

And was trying to use BulkMutation for writing into bigtable and was trying to set my own timestamp to set in the Mutation cell but getting 'Some mutations failed to apply'.
This is happening with all the version of "bulkMutate" functions.
I tried this on Bigtable-Emulator.

Below is my sample code:
public void writeToTableInBulk11() {

	try {

		System.out.println("\nWriting some greetings to the table");

		String[] greetings = { "Hello World!", "Hello Bigtable!", "Hello Java!" };



		BulkMutation batch = BulkMutation.create(tableId);

		for (int i = 0; i < greetings.length; i++) {

			//Working

			batch.add(ROW_KEY_PREFIX + i, Mutation.create().setCell(COLUMN_FAMILY, COLUMN_QUALIFIER, greetings[i]));

			//Not working

			batch.add(ROW_KEY_PREFIX + i, Mutation.create().setCell(COLUMN_FAMILY, COLUMN_QUALIFIER, System.currentTimeMillis(), greetings[i]));

		}

		//Tried both

// dataClient.bulkMutateRows(batch);

		dataClient.bulkMutationCallable().call(batch);



	} catch (NotFoundException e) {

		System.err.println("Failed to write to non-existent table: " + e.getMessage());

	}

}



public void writeToTableInBulk1() {

	try {

		System.out.println("\nWriting some greetings to the table");

		String[] greetings = { "Hello World!", "Hello Bigtable!", "Hello Java!" };



		BulkMutation batch = BulkMutation.create(tableId);

		for (int i = 0; i < greetings.length; i++) {

			//Not working

			batch.add(ROW_KEY_PREFIX + i, Mutation.create().setCell(COLUMN_FAMILY, COLUMN_QUALIFIER, System.currentTimeMillis(), greetings[i]));

			//Works

// batch.add(ROW_KEY_PREFIX + i, Mutation.create().setCell(COLUMN_FAMILY, COLUMN_QUALIFIER, greetings[i]));

		}

		ApiFuture<Void> resultFuture = dataClient.bulkMutateRowsAsync(batch);



		ApiFutures.addCallback(resultFuture, new ApiFutureCallback<Void>() {

			public void onFailure(Throwable t) {

				if (t instanceof BulkMutationFailure) {

					System.out.println("Some entries failed to apply");

				} else {

					t.printStackTrace();

				}

			}



			public void onSuccess(Void ignored) {

				System.out.println("Successfully applied all mutation");

			}

		}, MoreExecutors.directExecutor());



	} catch (NotFoundException e) {

		System.err.println("Failed to write to non-existent table: " + e.getMessage());

	}

}

Exception trace is below

Exception in thread "main" com.google.cloud.bigtable.data.v2.models.MutateRowsException: Some mutations failed to apply

at com.google.cloud.bigtable.data.v2.stub.mutaterows.MutateRowsAttemptCallable.handleAttemptSuccess(MutateRowsAttemptCallable.java:290)

at com.google.cloud.bigtable.data.v2.stub.mutaterows.MutateRowsAttemptCallable.access$000(MutateRowsAttemptCallable.java:87)

at com.google.cloud.bigtable.data.v2.stub.mutaterows.MutateRowsAttemptCallable$2.apply(MutateRowsAttemptCallable.java:121)

at com.google.cloud.bigtable.data.v2.stub.mutaterows.MutateRowsAttemptCallable$2.apply(MutateRowsAttemptCallable.java:118)

at com.google.api.core.ApiFutures$GaxFunctionToGuavaFunction.apply(ApiFutures.java:204)

at com.google.common.util.concurrent.AbstractTransformFuture$TransformFuture.doTransform(AbstractTransformFuture.java:243)

at com.google.common.util.concurrent.AbstractTransformFuture$TransformFuture.doTransform(AbstractTransformFuture.java:233)

at com.google.common.util.concurrent.AbstractTransformFuture.run(AbstractTransformFuture.java:118)

at com.google.common.util.concurrent.DirectExecutor.execute(DirectExecutor.java:30)

at com.google.common.util.concurrent.AbstractFuture.executeListener(AbstractFuture.java:1139)

at com.google.common.util.concurrent.AbstractFuture.complete(AbstractFuture.java:958)

at com.google.common.util.concurrent.AbstractFuture.set(AbstractFuture.java:726)

at com.google.common.util.concurrent.AbstractCatchingFuture.run(AbstractCatchingFuture.java:93)

at com.google.common.util.concurrent.DirectExecutor.execute(DirectExecutor.java:30)

at com.google.common.util.concurrent.AbstractFuture.executeListener(AbstractFuture.java:1139)

at com.google.common.util.concurrent.AbstractFuture.complete(AbstractFuture.java:958)

at com.google.common.util.concurrent.AbstractFuture.set(AbstractFuture.java:726)

at com.google.api.core.AbstractApiFuture$InternalSettableFuture.set(AbstractApiFuture.java:90)

at com.google.api.core.AbstractApiFuture.set(AbstractApiFuture.java:73)

at com.google.api.gax.rpc.SpoolingResponseObserver$MyFuture.set(SpoolingResponseObserver.java:84)

at com.google.api.gax.rpc.SpoolingResponseObserver.onCompleteImpl(SpoolingResponseObserver.java:72)

at com.google.api.gax.rpc.StateCheckingResponseObserver.onComplete(StateCheckingResponseObserver.java:74)

at com.google.api.gax.tracing.TracedResponseObserver.onComplete(TracedResponseObserver.java:109)

at com.google.api.gax.grpc.ExceptionResponseObserver.onCompleteImpl(ExceptionResponseObserver.java:89)

at com.google.api.gax.rpc.StateCheckingResponseObserver.onComplete(StateCheckingResponseObserver.java:74)

at com.google.api.gax.grpc.GrpcDirectStreamController$ResponseObserverAdapter.onClose(GrpcDirectStreamController.java:144)

at io.grpc.PartialForwardingClientCallListener.onClose(PartialForwardingClientCallListener.java:39)

at io.grpc.ForwardingClientCallListener.onClose(ForwardingClientCallListener.java:23)

at io.grpc.ForwardingClientCallListener$SimpleForwardingClientCallListener.onClose(ForwardingClientCallListener.java:40)

at io.grpc.internal.CensusStatsModule$StatsClientInterceptor$1$1.onClose(CensusStatsModule.java:700)

at io.grpc.PartialForwardingClientCallListener.onClose(PartialForwardingClientCallListener.java:39)

at io.grpc.ForwardingClientCallListener.onClose(ForwardingClientCallListener.java:23)

at io.grpc.ForwardingClientCallListener$SimpleForwardingClientCallListener.onClose(ForwardingClientCallListener.java:40)

at io.grpc.internal.CensusTracingModule$TracingClientInterceptor$1$1.onClose(CensusTracingModule.java:399)

at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:500)

at io.grpc.internal.ClientCallImpl.access$300(ClientCallImpl.java:65)

at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl.close(ClientCallImpl.java:592)

at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl.access$700(ClientCallImpl.java:508)

at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:632)

at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)

at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:123)

at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)

at java.util.concurrent.FutureTask.run(FutureTask.java:266)

at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)

at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)

at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)

at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)

at java.lang.Thread.run(Thread.java:748)

Suppressed: com.google.api.gax.rpc.AsyncTaskException: Asynchronous task failed

	at com.google.api.gax.rpc.ApiExceptions.callAndTranslateApiException(ApiExceptions.java:57)

	at com.google.api.gax.rpc.UnaryCallable.call(UnaryCallable.java:112)

	at ca.loblaw.kafka.connect.bigtable.util.HelloWorld.writeToTableInBulk11(HelloWorld.java:114)

	at ca.loblaw.kafka.connect.bigtable.util.HelloWorld.run(HelloWorld.java:81)

	at ca.loblaw.kafka.connect.bigtable.util.HelloWorld.main(HelloWorld.java:52)

[santy6781]

Hey, anyone looking into above issue?

[rahulKQL]

@santy6781 Apologies for the delay.

The timestamp value in Mutation#setCell must be a multiple of 1000 (millisecond granularity). For example: 1571848706482000.

I would suggest you to have a look at the native Bigtable client, which can be found in this repository, you may refer the documentation for the same here. This repository is primary focused on hbase wrappers.

[santy6781]

@rahulKQL thanks for the update.
I have two questions

1. Why this is not mention in any of the documentation? Won't it be great to have this documented?

The timestamp value in Mutation#setCell must be a multiple of 1000 (millisecond granularity). For example: 1571848706482000.

Any possibility if this will be fixed or added as feature in near future?

Bigtable Hbase API supports any valid timestamp value.

2. Are you suggesting that instead of using

<groupId>com.google.cloud.bigtable</groupId>

<artifactId>bigtable-client-core</artifactId>

<version>1.12.0</version>

use

<dependency>
  <groupId>com.google.cloud</groupId>
  <artifactId>google-cloud-bigtable</artifactId>
  <version>1.5.0</version>
</dependency>

If so, will it be possible for you to provide some of the major benefits and improvements (in terms of performance or any other) one will be getting using later one.

Thanks & Regards

[rahulKQL]

Here is my attempt to answer your queries

1. Why this is not mention in any of the documentation? Won't it be great to have this documented?

This was an overlook from our side, I have submitted PR#39 to address the same.

Any possibility if this will be fixed or added as feature in near future?

This is intentional behavior and is consistent across all native(java/ruby/python etc.) clients.

Bigtable Hbase API supports any valid timestamp value.

This is because "Bigtable timestamp has more granularity than HBase one", bigtable hbase client dedup the timestamp and converts it to microseconds before sending it to server.

2. Are you suggesting that instead of using

Yes, I am suggesting "google-cloud-bigtable" over this client.

If so, will it be possible for you to provide some of the major benefits and improvements (in terms of performance or any other) one will be getting using later one.

The native client exposes user-friendly operation along with callables to support advanced use cases. It also guarantees consistent behavior across language.
Currently, benchmark test cases are in progress, but mutating rows using native client is twice the performant then bigtable-client core.

[santy6781]

@rahulKQL Thanks for the clarification, really appreciate.

[rjgrover13]

I was getting the same issue and looked every where for the solution which didn't work for me as I've already checked the timestamp and column family are same. As per the documentation the mutations failed if the size is more than 100k. But in my case, I was processing 46k. So to debug the issue, I tried it first by doing it rowMutate() one by one instead of bulk mutation and it worked but that's very time consuming so then I created my own batch List<BulkMutation> batchBulkRowMutation = new ArrayList<>(); and then processing those batch as in bulk mutation.

`
List batchBulkRowMutation = new ArrayList<>();
RowMutationEntry rowMutation = null;
long btRowCounter = 0;
try {
BulkMutation bulkMutation = BulkMutation.create(bigTableId);
for (Row row : dispatchPlanRowStream) { // reading data from other bigtable
var rowKey = row.getKey().toStringUtf8();

            if (rowKey != null) {
                rowMutation = RowMutationEntry.create(rowKey);
                rowMutation = addingToBT(rowKey, rowMutation, row); // setting cell data for the column family in separate method, this is simple you can refer this in docs (calling setCell method and adding column family, qualifier, timestamp and value to it).
                bulkMutation.add(rowMutation);
            }
            btRowCounter++;

            if (btRowCounter % 20000 == 0) { // creating batch for 20k records
                batchBulkRowMutation.add(bulkMutation); // adding to the list
                bulkMutation = BulkMutation.create(bigTableId); // reassigning the value into bulkMutation to get other 20k records
            }
        }


        if (rowMutation != null) {
                // bulk mutation
            BigtableDataClient dataClient = GCPBigtableConnection.getBigtableDataClient(gcpProjectId, bigTableInstanceId);
            dataClient.bulkMutateRows(bulkMutation); // first processing the odd one records, let assume we have 50k records so for that we created 2 batch and the rest of records will be processed here or you can add those other records into the list as well to optimize the code.
      
            if (!batchBulkRowMutation.isEmpty()) {
                for (int j =0; j < batchBulkRowMutation.size(); j++) {
                    dataClient.bulkMutateRows(batchBulkRowMutation.get(j)); // processing the batch over here iterating the list of bulk mutations
                }
            }
        }
    } catch (Exception ex) {
        log.error("Failed to write to BigTable process"); // print your own exception
    }

`

Not sure if there is any ideal solution but didn't find anything on this. Hope this will help, thanks