Skip to content

Fix error handling for mutations in Cassandra #2827

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Jul 8, 2025
Merged

Fix error handling for mutations in Cassandra #2827

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
38311d2
Fix error handling for mutations in Cassandra
brfrn169 Jul 6, 2025
59587f1
Fix based on feedback
brfrn169 Jul 6, 2025
bab68f4
Merge branch 'master' into fix-error-handling-for-mutations-in-cassandra
brfrn169 Jul 8, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 8 additions & 8 deletions core/src/main/java/com/scalar/db/storage/cassandra/BatchHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,8 +11,8 @@
import com.google.common.annotations.VisibleForTesting;
import com.scalar.db.api.Mutation;
import com.scalar.db.common.CoreError;
import com.scalar.db.exception.storage.ExecutionException;
import com.scalar.db.exception.storage.NoMutationException;
import com.scalar.db.exception.storage.RetriableExecutionException;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import java.util.List;
import javax.annotation.concurrent.ThreadSafe;
Expand Down Expand Up @@ -48,12 +48,11 @@ public BatchHandler(Session session, StatementHandlerManager handlers) {
* must be for the same partition.
*
* @param mutations a list of {@code Mutation}s to execute
* @throws RetriableExecutionException if it fails, but it can be retried
* @throws NoMutationException if at least one of conditional {@code Mutation}s fails because it
* didn't meet the condition
* @throws ExecutionException if the execution fails
*/
public void handle(List<? extends Mutation> mutations)
throws RetriableExecutionException, NoMutationException {
public void handle(List<? extends Mutation> mutations) throws ExecutionException {
try {
ResultSet results = execute(mutations);
// it's for conditional update. non-conditional update always return true
Expand All @@ -64,16 +63,17 @@ public void handle(List<? extends Mutation> mutations)
logger.warn("Write timeout happened during batch mutate operation", e);
WriteType writeType = e.getWriteType();
if (writeType == WriteType.BATCH_LOG) {
throw new RetriableExecutionException(
CoreError.CASSANDRA_LOGGING_FAILED_IN_BATCH.buildMessage(), e);
// A timeout occurred while the coordinator was waiting for the batch log replicas to
// acknowledge the log. Thus, the batch may or may not be applied.
throw new ExecutionException(CoreError.CASSANDRA_LOGGING_FAILED_IN_BATCH.buildMessage(), e);
} else if (writeType == WriteType.BATCH) {
logger.warn("Logging succeeded, but mutations in the batch partially failed", e);
} else {
throw new RetriableExecutionException(
throw new ExecutionException(
CoreError.CASSANDRA_OPERATION_FAILED_IN_BATCH.buildMessage(writeType), e);
}
} catch (RuntimeException e) {
throw new RetriableExecutionException(
throw new ExecutionException(
CoreError.CASSANDRA_ERROR_OCCURRED_IN_BATCH.buildMessage(e.getMessage()), e);
}
}
Expand Down
11 changes: 4 additions & 7 deletions core/src/main/java/com/scalar/db/storage/cassandra/MutateStatementHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,6 @@
import com.scalar.db.common.CoreError;
import com.scalar.db.exception.storage.ExecutionException;
import com.scalar.db.exception.storage.NoMutationException;
import com.scalar.db.exception.storage.RetriableExecutionException;
import javax.annotation.Nonnull;
import javax.annotation.concurrent.ThreadSafe;

Expand All @@ -28,9 +27,9 @@ public MutateStatementHandler(Session session) {
*
* @param operation {@link Mutation} operation
* @return a {@code ResultSet}
* @throws RetriableExecutionException if the execution fails, but it can be retriable
* @throws ReadRepairableExecutionException if the execution partially fails, which can be
* repaired by a following read
* @throws ExecutionException if the execution fails
*/
@Override
@Nonnull
Expand All @@ -45,8 +44,7 @@ public ResultSet handle(Operation operation) throws ExecutionException {
return results;
} catch (WriteTimeoutException e) {
if (e.getWriteType() == WriteType.CAS) {
// retry needs to be done if applications need to do the operation exactly
throw new RetriableExecutionException(
throw new ExecutionException(
CoreError.CASSANDRA_WRITE_TIMEOUT_IN_PAXOS_PHASE_IN_MUTATION.buildMessage(), e);
} else if (e.getWriteType() == WriteType.SIMPLE) {
Mutation mutation = (Mutation) operation;
Expand All @@ -55,8 +53,7 @@ public ResultSet handle(Operation operation) throws ExecutionException {
throw new ReadRepairableExecutionException(
CoreError.CASSANDRA_WRITE_TIMEOUT_IN_LEARN_PHASE_IN_MUTATION.buildMessage(), e);
} else {
// retry needs to be done if applications need to do the operation exactly
throw new RetriableExecutionException(
throw new ExecutionException(
CoreError.CASSANDRA_WRITE_TIMEOUT_SIMPLE_WRITE_OPERATION_FAILED_IN_MUTATION
.buildMessage(),
e);
Expand All @@ -66,7 +63,7 @@ public ResultSet handle(Operation operation) throws ExecutionException {
CoreError.CASSANDRA_WRITE_TIMEOUT_WITH_OTHER_WRITE_TYPE_IN_MUTATION.buildMessage(), e);
}
} catch (RuntimeException e) {
throw new RetriableExecutionException(
throw new ExecutionException(
CoreError.CASSANDRA_ERROR_OCCURRED_IN_MUTATION.buildMessage(e.getMessage()), e);
}
}
Expand Down
19 changes: 9 additions & 10 deletions core/src/test/java/com/scalar/db/storage/cassandra/BatchHandlerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,8 @@
import com.scalar.db.api.Put;
import com.scalar.db.api.PutIfExists;
import com.scalar.db.api.PutIfNotExists;
import com.scalar.db.exception.storage.ExecutionException;
import com.scalar.db.exception.storage.NoMutationException;
import com.scalar.db.exception.storage.RetriableExecutionException;
import com.scalar.db.io.Key;
import java.util.Arrays;
import java.util.List;
Expand Down Expand Up @@ -178,7 +178,7 @@ public void handle_CorrectHandlerAndAtLeastOneConditionalPutGiven_ShouldSetConsi
}

@Test
public void handle_WTEThrownInLoggingInBatchExecution_ShouldThrowRetriableExecutionException() {
public void handle_WTEThrownInLoggingInBatchExecution_ShouldThrowExecutionException() {
// Arrange
configureBehavior();
mutations = prepareConditionalPuts();
Expand All @@ -188,7 +188,7 @@ public void handle_WTEThrownInLoggingInBatchExecution_ShouldThrowRetriableExecut

// Act Assert
assertThatThrownBy(() -> batch.handle(mutations))
.isInstanceOf(RetriableExecutionException.class)
.isInstanceOf(ExecutionException.class)
.hasCause(e);
}

Expand All @@ -206,7 +206,7 @@ public void handle_WTEThrownInMutationInBatchExecution_ShouldExecuteProperly() {
}

@Test
public void handle_WTEThrownInCasInBatchExecution_ShouldThrowRetriableExecutionException() {
public void handle_WTEThrownInCasInBatchExecution_ShouldThrowExecutionException() {
// Arrange
configureBehavior();
mutations = prepareConditionalPuts();
Expand All @@ -216,13 +216,12 @@ public void handle_WTEThrownInCasInBatchExecution_ShouldThrowRetriableExecutionE

// Act Assert
assertThatThrownBy(() -> batch.handle(mutations))
.isInstanceOf(RetriableExecutionException.class)
.isInstanceOf(ExecutionException.class)
.hasCause(e);
}

@Test
public void
handle_WTEThrownInSimpleWriteInBatchExecution_ShouldThrowRetriableExecutionException() {
public void handle_WTEThrownInSimpleWriteInBatchExecution_ShouldThrowExecutionException() {
// Arrange
configureBehavior();
mutations = prepareConditionalPuts();
Expand All @@ -232,12 +231,12 @@ public void handle_WTEThrownInCasInBatchExecution_ShouldThrowRetriableExecutionE

// Act Assert
assertThatThrownBy(() -> batch.handle(mutations))
.isInstanceOf(RetriableExecutionException.class)
.isInstanceOf(ExecutionException.class)
.hasCause(e);
}

@Test
public void handle_DriverExceptionThrownInExecution_ShouldThrowRetriableExecutionException() {
public void handle_DriverExceptionThrownInExecution_ShouldThrowExecutionException() {
// Arrange
configureBehavior();
mutations = prepareConditionalPuts();
Expand All @@ -246,7 +245,7 @@ public void handle_DriverExceptionThrownInExecution_ShouldThrowRetriableExecutio

// Act Assert
assertThatThrownBy(() -> batch.handle(mutations))
.isInstanceOf(RetriableExecutionException.class)
.isInstanceOf(ExecutionException.class)
.hasCause(e);
}

Expand Down
14 changes: 6 additions & 8 deletions core/src/test/java/com/scalar/db/storage/cassandra/InsertStatementHandlerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,6 @@
import com.scalar.db.api.PutIfNotExists;
import com.scalar.db.exception.storage.ExecutionException;
import com.scalar.db.exception.storage.NoMutationException;
import com.scalar.db.exception.storage.RetriableExecutionException;
import com.scalar.db.io.Key;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
Expand Down Expand Up @@ -330,7 +329,7 @@ public void setConsistency_PutOperationWithIfNotExistsGiven_ShouldPrepareWithQuo
}

@Test
public void handle_WTEWithCasThrown_ShouldThrowProperExecutionException() {
public void handle_WTEWithCasThrown_ShouldThrowExecutionException() {
// Arrange
put = preparePutWithClusteringKey();
put.withCondition(new PutIfNotExists());
Expand All @@ -342,13 +341,12 @@ public void handle_WTEWithCasThrown_ShouldThrowProperExecutionException() {

// Act Assert
assertThatThrownBy(() -> spy.handle(put))
.isInstanceOf(RetriableExecutionException.class)
.isInstanceOf(ExecutionException.class)
.hasCause(toThrow);
}

@Test
public void
handle_PutWithConditionGivenAndWTEWithSimpleThrown_ShouldThrowProperExecutionException() {
public void handle_PutWithConditionGivenAndWTEWithSimpleThrown_ShouldThrowExecutionException() {
// Arrange
put = preparePutWithClusteringKey();
put.withCondition(new PutIfNotExists());
Expand All @@ -366,7 +364,7 @@ public void handle_WTEWithCasThrown_ShouldThrowProperExecutionException() {

@Test
public void
handle_PutWithoutConditionGivenAndWTEWithSimpleThrown_ShouldThrowProperExecutionException() {
handle_PutWithoutConditionGivenAndWTEWithSimpleThrown_ShouldThrowExecutionException() {
// Arrange
put = preparePutWithClusteringKey();
spy = prepareSpiedInsertStatementHandler();
Expand All @@ -377,12 +375,12 @@ public void handle_WTEWithCasThrown_ShouldThrowProperExecutionException() {

// Act Assert
assertThatThrownBy(() -> spy.handle(put))
.isInstanceOf(RetriableExecutionException.class)
.isInstanceOf(ExecutionException.class)
.hasCause(toThrow);
}

@Test
public void handle_DriverExceptionThrown_ShouldThrowProperExecutionException() {
public void handle_DriverExceptionThrown_ShouldThrowExecutionException() {
// Arrange
put = preparePutWithClusteringKey();
spy = prepareSpiedInsertStatementHandler();
Expand Down