BulkMutation failure: MutateRowsException: Some mutations failed to apply

[santosh sharma]

Dear Solomon/Bigtable-Team,

I am working on bigtable java api

<dependency>

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

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

    <version>1.12.0</version>

</dependency>

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++) {

//Not working

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

//Works

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)

Is there a way to fix this issue or it is a known issue.

--

Thanks & Regards
Santosh