Improve logging in Consensus Commit (#2915)

Author: brfrn169

core/src/main/java/com/scalar/db/common/CoreError.java

@@ -728,19 +728,35 @@ public enum CoreError implements ScalarDbError {
       "",
       ""),
   CONSENSUS_COMMIT_PREPARING_RECORD_EXISTS(
-      Category.CONCURRENCY_ERROR, "0013", "The record being prepared already exists", "", ""),
+      Category.CONCURRENCY_ERROR,
+      "0013",
+      "The record being prepared already exists. Details: %s",
+      "",
+      ""),
   CONSENSUS_COMMIT_CONFLICT_OCCURRED_WHEN_PREPARING_RECORDS(
-      Category.CONCURRENCY_ERROR, "0014", "A conflict occurred when preparing records", "", ""),
+      Category.CONCURRENCY_ERROR,
+      "0014",
+      "A conflict occurred when preparing records. Details: %s",
+      "",
+      ""),
   CONSENSUS_COMMIT_CONFLICT_OCCURRED_WHEN_COMMITTING_STATE(
       Category.CONCURRENCY_ERROR,
       "0015",
-      "The committing state in the coordinator failed. The transaction has been aborted",
+      "The committing state in the coordinator failed. The transaction has been aborted. Details: %s",
       "",
       ""),
   CONSENSUS_COMMIT_CONFLICT_OCCURRED_WHILE_IMPLICIT_PRE_READ(
-      Category.CONCURRENCY_ERROR, "0016", "A conflict occurred during implicit pre-read", "", ""),
+      Category.CONCURRENCY_ERROR,
+      "0016",
+      "A conflict occurred during implicit pre-read. Details: %s",
+      "",
+      ""),
   CONSENSUS_COMMIT_READ_UNCOMMITTED_RECORD(
-      Category.CONCURRENCY_ERROR, "0017", "This record needs to be recovered", "", ""),
+      Category.CONCURRENCY_ERROR,
+      "0017",
+      "This record needs to be recovered. Table: %s; Partition Key: %s; Clustering Key: %s; Transaction ID that wrote the record: %s",
+      "",
+      ""),
   CONSENSUS_COMMIT_CONDITION_NOT_SATISFIED_BECAUSE_RECORD_NOT_EXISTS(
       Category.CONCURRENCY_ERROR,
       "0018",
@@ -784,7 +800,11 @@ public enum CoreError implements ScalarDbError {
       "",
       ""),
   CONSENSUS_COMMIT_CONFLICT_OCCURRED_WHEN_COMMITTING_RECORDS(
-      Category.CONCURRENCY_ERROR, "0026", "A conflict occurred when committing records", "", ""),
+      Category.CONCURRENCY_ERROR,
+      "0026",
+      "A conflict occurred when committing records. Details: %s",
+      "",
+      ""),
 
   //
   // Errors for the internal error category
@@ -890,20 +910,21 @@ public enum CoreError implements ScalarDbError {
   JDBC_TRANSACTION_BEGINNING_TRANSACTION_FAILED(
       Category.INTERNAL_ERROR, "0035", "Beginning a transaction failed. Details: %s", "", ""),
   CONSENSUS_COMMIT_PREPARING_RECORDS_FAILED(
-      Category.INTERNAL_ERROR, "0036", "Preparing records failed", "", ""),
-  CONSENSUS_COMMIT_VALIDATION_FAILED(Category.INTERNAL_ERROR, "0037", "Validation failed", "", ""),
+      Category.INTERNAL_ERROR, "0036", "Preparing records failed. Details: %s", "", ""),
+  CONSENSUS_COMMIT_VALIDATION_FAILED(
+      Category.INTERNAL_ERROR, "0037", "Validation failed. Details: %s", "", ""),
   CONSENSUS_COMMIT_EXECUTING_IMPLICIT_PRE_READ_FAILED(
-      Category.INTERNAL_ERROR, "0038", "Executing implicit pre-read failed", "", ""),
+      Category.INTERNAL_ERROR, "0038", "Executing implicit pre-read failed. Details: %s", "", ""),
   CONSENSUS_COMMIT_READING_RECORD_FROM_STORAGE_FAILED(
       Category.INTERNAL_ERROR,
       "0039",
-      "Reading a record from the underlying storage failed",
+      "Reading a record from the underlying storage failed. Details: %s",
       "",
       ""),
   CONSENSUS_COMMIT_SCANNING_RECORDS_FROM_STORAGE_FAILED(
       Category.INTERNAL_ERROR,
       "0040",
-      "Scanning records from the underlying storage failed",
+      "Scanning records from the underlying storage failed. Details: %s",
       "",
       ""),
   CONSENSUS_COMMIT_ROLLBACK_FAILED_BECAUSE_TRANSACTION_ALREADY_COMMITTED(
@@ -919,7 +940,7 @@ public enum CoreError implements ScalarDbError {
       Category.INTERNAL_ERROR, "0044", "The Upsert operation failed. Details: %s", "", ""),
   JDBC_TRANSACTION_UPDATE_OPERATION_FAILED(
       Category.INTERNAL_ERROR, "0045", "The Update operation failed. Details: %s", "", ""),
-  HANDLING_BEFORE_PREPARATION_SNAPSHOT_HOOK_FAILED(
+  CONSENSUS_COMMIT_HANDLING_BEFORE_PREPARATION_SNAPSHOT_HOOK_FAILED(
       Category.INTERNAL_ERROR,
       "0046",
       "Handling the before-preparation snapshot hook failed. Details: %s",
@@ -938,7 +959,7 @@ public enum CoreError implements ScalarDbError {
   CONSENSUS_COMMIT_RECOVERING_RECORDS_FAILED(
       Category.INTERNAL_ERROR, "0057", "Recovering records failed. Details: %s", "", ""),
   CONSENSUS_COMMIT_COMMITTING_RECORDS_FAILED(
-      Category.INTERNAL_ERROR, "0058", "Committing records failed", "", ""),
+      Category.INTERNAL_ERROR, "0058", "Committing records failed. Details: %s", "", ""),
 
   //
   // Errors for the unknown transaction status error category
@@ -952,21 +973,31 @@ public enum CoreError implements ScalarDbError {
   CONSENSUS_COMMIT_COMMITTING_STATE_FAILED_WITH_NO_MUTATION_EXCEPTION_BUT_COORDINATOR_STATUS_DOES_NOT_EXIST(
       Category.UNKNOWN_TRANSACTION_STATUS_ERROR,
       "0001",
-      "Committing state failed with NoMutationException, but the coordinator status does not exist",
+      "Committing state failed with NoMutationException, but the coordinator status does not exist. Details: %s",
+      "",
+      ""),
+  CONSENSUS_COMMIT_CANNOT_COORDINATOR_STATUS(
+      Category.UNKNOWN_TRANSACTION_STATUS_ERROR,
+      "0002",
+      "The coordinator status cannot be retrieved. Details: %s",
       "",
       ""),
-  CONSENSUS_COMMIT_CANNOT_GET_STATE(
-      Category.UNKNOWN_TRANSACTION_STATUS_ERROR, "0002", "The state cannot be retrieved", "", ""),
   CONSENSUS_COMMIT_UNKNOWN_COORDINATOR_STATUS(
       Category.UNKNOWN_TRANSACTION_STATUS_ERROR,
       "0003",
-      "The coordinator status is unknown",
+      "The coordinator status is unknown. Details: %s",
       "",
       ""),
   CONSENSUS_COMMIT_ABORTING_STATE_FAILED_WITH_NO_MUTATION_EXCEPTION_BUT_COORDINATOR_STATUS_DOES_NOT_EXIST(
       Category.UNKNOWN_TRANSACTION_STATUS_ERROR,
       "0004",
-      "Aborting state failed with NoMutationException, but the coordinator status does not exist",
+      "Aborting state failed with NoMutationException, but the coordinator status does not exist. Details: %s",
+      "",
+      ""),
+  CONSENSUS_COMMIT_ONE_PHASE_COMMITTING_RECORDS_FAILED(
+      Category.UNKNOWN_TRANSACTION_STATUS_ERROR,
+      "0005",
+      "One-phase committing records failed. Details: %s",
       "",
       ""),
   ;

core/src/main/java/com/scalar/db/exception/storage/NoMutationException.java

@@ -1,12 +1,62 @@
 package com.scalar.db.exception.storage;
 
+import com.google.common.collect.ImmutableList;
+import com.scalar.db.api.Delete;
+import com.scalar.db.api.Mutation;
+import com.scalar.db.api.Put;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.util.List;
+
 public class NoMutationException extends ExecutionException {
 
-  public NoMutationException(String message) {
-    super(message);
+  private final List<? extends Mutation> mutations;
+
+  public NoMutationException(
+      String message, @SuppressFBWarnings("EI_EXPOSE_REP2") List<? extends Mutation> mutations) {
+    super(addTransactionIdToMessage(message, mutations));
+    this.mutations = mutations;
   }
 
-  public NoMutationException(String message, Throwable cause) {
-    super(message, cause);
+  public NoMutationException(
+      String message,
+      @SuppressFBWarnings("EI_EXPOSE_REP2") List<? extends Mutation> mutations,
+      Throwable cause) {
+    super(addTransactionIdToMessage(message, mutations), cause);
+    this.mutations = mutations;
+  }
+
+  public List<? extends Mutation> getMutations() {
+    return ImmutableList.copyOf(mutations);
+  }
+
+  private static String addTransactionIdToMessage(
+      String message, List<? extends Mutation> mutations) {
+    StringBuilder builder = new StringBuilder(message).append(". Mutations: [");
+
+    boolean first = true;
+    for (Mutation mutation : mutations) {
+      assert mutation.forFullTableName().isPresent();
+      assert mutation instanceof Put || mutation instanceof Delete;
+
+      if (!first) {
+        builder.append(", ");
+      } else {
+        first = false;
+      }
+      builder
+          .append("{Type: ")
+          .append(mutation.getClass().getSimpleName())
+          .append(", Table: ")
+          .append(mutation.forFullTableName().get())
+          .append(", Partition Key: ")
+          .append(mutation.getPartitionKey())
+          .append(", Clustering Key: ")
+          .append(mutation.getClusteringKey())
+          .append(", Condition: ")
+          .append(mutation.getCondition())
+          .append("}");
+    }
+
+    return builder.append("]").toString();
   }
 }

core/src/main/java/com/scalar/db/storage/cassandra/BatchHandler.java

@@ -57,7 +57,7 @@ public void handle(List<? extends Mutation> mutations) throws ExecutionException
       ResultSet results = execute(mutations);
       // it's for conditional update. non-conditional update always return true
       if (!results.wasApplied()) {
-        throw new NoMutationException(CoreError.NO_MUTATION_APPLIED.buildMessage());
+        throw new NoMutationException(CoreError.NO_MUTATION_APPLIED.buildMessage(), mutations);
       }
     } catch (WriteTimeoutException e) {
       logger.warn("Write timeout happened during batch mutate operation", e);

core/src/main/java/com/scalar/db/storage/cassandra/MutateStatementHandler.java

@@ -12,6 +12,7 @@
 import com.scalar.db.common.CoreError;
 import com.scalar.db.exception.storage.ExecutionException;
 import com.scalar.db.exception.storage.NoMutationException;
+import java.util.Collections;
 import javax.annotation.Nonnull;
 import javax.annotation.concurrent.ThreadSafe;
 
@@ -39,7 +40,8 @@ public ResultSet handle(Operation operation) throws ExecutionException {
 
       Mutation mutation = (Mutation) operation;
       if (mutation.getCondition().isPresent() && !results.one().getBool(0)) {
-        throw new NoMutationException(CoreError.NO_MUTATION_APPLIED.buildMessage());
+        throw new NoMutationException(
+            CoreError.NO_MUTATION_APPLIED.buildMessage(), Collections.singletonList(mutation));
       }
       return results;
     } catch (WriteTimeoutException e) {

core/src/main/java/com/scalar/db/storage/cosmos/BatchHandler.java

@@ -49,7 +49,7 @@ public void handle(List<? extends Mutation> mutations) throws ExecutionException
     try {
       executeStoredProcedure(mutations, tableMetadata);
     } catch (CosmosException e) {
-      throwException(e);
+      throwException(e, mutations);
     } catch (RuntimeException e) {
       throw new ExecutionException(
           CoreError.COSMOS_ERROR_OCCURRED_IN_MUTATION.buildMessage(e.getMessage()), e);
@@ -84,11 +84,13 @@ private void executeStoredProcedure(
         .execute(args, cosmosMutation.getStoredProcedureOptions());
   }
 
-  private void throwException(CosmosException exception) throws ExecutionException {
+  private void throwException(CosmosException exception, List<? extends Mutation> mutations)
+      throws ExecutionException {
     int statusCode = exception.getSubStatusCode();
 
     if (statusCode == CosmosErrorCode.PRECONDITION_FAILED.get()) {
-      throw new NoMutationException(CoreError.NO_MUTATION_APPLIED.buildMessage(), exception);
+      throw new NoMutationException(
+          CoreError.NO_MUTATION_APPLIED.buildMessage(), mutations, exception);
     } else if (statusCode == CosmosErrorCode.RETRY_WITH.get()) {
       throw new RetriableExecutionException(
           CoreError.COSMOS_RETRY_WITH_ERROR_OCCURRED_IN_MUTATION.buildMessage(

core/src/main/java/com/scalar/db/storage/cosmos/MutateStatementHandler.java

@@ -10,6 +10,7 @@
 import com.scalar.db.exception.storage.NoMutationException;
 import com.scalar.db.exception.storage.RetriableExecutionException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import javax.annotation.concurrent.ThreadSafe;
 
@@ -32,7 +33,7 @@ public void handle(Mutation mutation) throws ExecutionException {
     try {
       execute(mutation);
     } catch (CosmosException e) {
-      throwException(e);
+      throwException(e, mutation);
     } catch (RuntimeException e) {
       throw new ExecutionException(
           CoreError.COSMOS_ERROR_OCCURRED_IN_MUTATION.buildMessage(e.getMessage()), e);
@@ -56,11 +57,15 @@ protected void executeStoredProcedure(Mutation mutation, TableMetadata tableMeta
         .execute(args, cosmosMutation.getStoredProcedureOptions());
   }
 
-  private void throwException(CosmosException exception) throws ExecutionException {
+  private void throwException(CosmosException exception, Mutation mutation)
+      throws ExecutionException {
     int statusCode = exception.getSubStatusCode();
 
     if (statusCode == CosmosErrorCode.PRECONDITION_FAILED.get()) {
-      throw new NoMutationException(CoreError.NO_MUTATION_APPLIED.buildMessage(), exception);
+      throw new NoMutationException(
+          CoreError.NO_MUTATION_APPLIED.buildMessage(),
+          Collections.singletonList(mutation),
+          exception);
     } else if (statusCode == CosmosErrorCode.RETRY_WITH.get()) {
       throw new RetriableExecutionException(
           CoreError.COSMOS_RETRY_WITH_ERROR_OCCURRED_IN_MUTATION.buildMessage(

core/src/main/java/com/scalar/db/storage/dynamo/BatchHandler.java

@@ -86,7 +86,7 @@ public void handle(List<? extends Mutation> mutations) throws ExecutionException
       boolean allReasonsAreTransactionConflicts = true;
       for (CancellationReason reason : e.cancellationReasons()) {
         if (reason.code().equals("ConditionalCheckFailed")) {
-          throw new NoMutationException(CoreError.NO_MUTATION_APPLIED.buildMessage(), e);
+          throw new NoMutationException(CoreError.NO_MUTATION_APPLIED.buildMessage(), mutations, e);
         }
         if (!reason.code().equals("TransactionConflict") && !reason.code().equals("None")) {
           allReasonsAreTransactionConflicts = false;

core/src/main/java/com/scalar/db/storage/dynamo/DeleteStatementHandler.java

@@ -12,6 +12,7 @@
 import com.scalar.db.exception.storage.NoMutationException;
 import com.scalar.db.exception.storage.RetriableExecutionException;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.util.Collections;
 import java.util.Map;
 import java.util.Optional;
 import javax.annotation.concurrent.ThreadSafe;
@@ -50,7 +51,8 @@ public void handle(Delete delete) throws ExecutionException {
     try {
       delete(delete, tableMetadata);
     } catch (ConditionalCheckFailedException e) {
-      throw new NoMutationException(CoreError.NO_MUTATION_APPLIED.buildMessage(), e);
+      throw new NoMutationException(
+          CoreError.NO_MUTATION_APPLIED.buildMessage(), Collections.singletonList(delete), e);
     } catch (TransactionConflictException e) {
       throw new RetriableExecutionException(
           CoreError.DYNAMO_TRANSACTION_CONFLICT_OCCURRED_IN_MUTATION.buildMessage(

core/src/main/java/com/scalar/db/storage/dynamo/PutStatementHandler.java

@@ -13,6 +13,7 @@
 import com.scalar.db.exception.storage.NoMutationException;
 import com.scalar.db.exception.storage.RetriableExecutionException;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.util.Collections;
 import java.util.Map;
 import java.util.Optional;
 import javax.annotation.concurrent.ThreadSafe;
@@ -51,7 +52,8 @@ public void handle(Put put) throws ExecutionException {
     try {
       execute(put, tableMetadata);
     } catch (ConditionalCheckFailedException e) {
-      throw new NoMutationException(CoreError.NO_MUTATION_APPLIED.buildMessage(), e);
+      throw new NoMutationException(
+          CoreError.NO_MUTATION_APPLIED.buildMessage(), Collections.singletonList(put), e);
     } catch (TransactionConflictException e) {
       throw new RetriableExecutionException(
           CoreError.DYNAMO_TRANSACTION_CONFLICT_OCCURRED_IN_MUTATION.buildMessage(

core/src/main/java/com/scalar/db/storage/jdbc/JdbcDatabase.java

@@ -21,6 +21,7 @@
 import com.scalar.db.exception.storage.RetriableExecutionException;
 import java.sql.Connection;
 import java.sql.SQLException;
+import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 import javax.annotation.concurrent.ThreadSafe;
@@ -138,7 +139,8 @@ public void put(Put put) throws ExecutionException {
     try {
       connection = dataSource.getConnection();
       if (!jdbcService.put(put, connection)) {
-        throw new NoMutationException(CoreError.NO_MUTATION_APPLIED.buildMessage());
+        throw new NoMutationException(
+            CoreError.NO_MUTATION_APPLIED.buildMessage(), Collections.singletonList(put));
       }
     } catch (SQLException e) {
       throw new ExecutionException(
@@ -160,7 +162,8 @@ public void delete(Delete delete) throws ExecutionException {
     try {
       connection = dataSource.getConnection();
       if (!jdbcService.delete(delete, connection)) {
-        throw new NoMutationException(CoreError.NO_MUTATION_APPLIED.buildMessage());
+        throw new NoMutationException(
+            CoreError.NO_MUTATION_APPLIED.buildMessage(), Collections.singletonList(delete));
       }
     } catch (SQLException e) {
       throw new ExecutionException(
@@ -210,7 +213,7 @@ public void mutate(List<? extends Mutation> mutations) throws ExecutionException
           throw new ExecutionException(
               CoreError.JDBC_ERROR_OCCURRED_IN_MUTATION.buildMessage(e.getMessage()), e);
         }
-        throw new NoMutationException(CoreError.NO_MUTATION_APPLIED.buildMessage());
+        throw new NoMutationException(CoreError.NO_MUTATION_APPLIED.buildMessage(), mutations);
       } else {
         connection.commit();
       }