diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/readrows/StateMachine.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/readrows/StateMachine.java
index 5703dc9a76..b6b6db678f 100644
--- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/readrows/StateMachine.java
+++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/readrows/StateMachine.java
@@ -18,7 +18,9 @@
 import com.google.bigtable.v2.ReadRowsResponse.CellChunk;
 import com.google.cloud.bigtable.data.v2.internal.ByteStringComparator;
 import com.google.cloud.bigtable.data.v2.models.RowAdapter.RowBuilder;
+import com.google.common.base.Joiner;
 import com.google.common.base.Preconditions;
+import com.google.common.collect.EvictingQueue;
 import com.google.protobuf.ByteString;
 import java.util.List;
 
@@ -77,6 +79,14 @@ final class StateMachine<RowT> {
   private State currentState;
   private ByteString lastCompleteRowKey;
 
+  // debug stats
+  private int numScannedNotifications = 0;
+  private int numRowsCommitted = 0;
+  private int numChunksProcessed = 0;
+  private int numCellsInRow = 0;
+  private int numCellsInLastRow = 0;
+  private EvictingQueue<ByteString> lastSeenKeys = EvictingQueue.create(5);
+
   // Track current cell attributes: protocol omits them when they are repeated
   private ByteString rowKey;
   private String familyName;
@@ -120,6 +130,7 @@ final class StateMachine<RowT> {
    */
   void handleLastScannedRow(ByteString key) {
     try {
+      numScannedNotifications++;
       currentState = currentState.handleLastScannedRow(key);
     } catch (RuntimeException e) {
       currentState = null;
@@ -148,6 +159,7 @@ void handleLastScannedRow(ByteString key) {
    */
   void handleChunk(CellChunk chunk) {
     try {
+      numChunksProcessed++;
       currentState = currentState.handleChunk(chunk);
     } catch (RuntimeException e) {
       currentState = null;
@@ -191,6 +203,7 @@ private void reset() {
     expectedCellSize = 0;
     remainingCellBytes = 0;
     completeRow = null;
+    numCellsInRow = 0;
 
     adapter.reset();
   }
@@ -326,6 +339,7 @@ State handleChunk(CellChunk chunk) {
             return AWAITING_CELL_VALUE;
           }
           adapter.finishCell();
+          numCellsInRow++;
 
           if (!chunk.getCommitRow()) {
             return AWAITING_NEW_CELL;
@@ -374,6 +388,7 @@ State handleChunk(CellChunk chunk) {
             return AWAITING_CELL_VALUE;
           }
           adapter.finishCell();
+          numCellsInRow++;
 
           if (!chunk.getCommitRow()) {
             return AWAITING_NEW_CELL;
@@ -416,12 +431,31 @@ private State handleCommit() {
     validate(remainingCellBytes == 0, "Can't commit with remaining bytes");
     completeRow = adapter.finishRow();
     lastCompleteRowKey = rowKey;
+
+    lastSeenKeys.add(rowKey);
+    numRowsCommitted++;
+    numCellsInLastRow = numCellsInRow;
     return AWAITING_ROW_CONSUME;
   }
 
-  private static void validate(boolean condition, String message) {
+  private void validate(boolean condition, String message) {
     if (!condition) {
-      throw new InvalidInputException(message);
+      throw new InvalidInputException(
+          message
+              + ". numScannedNotifications: "
+              + numScannedNotifications
+              + ", numRowsCommitted: "
+              + numRowsCommitted
+              + ", numChunksProcessed: "
+              + numChunksProcessed
+              + ", numCellsInRow: "
+              + numCellsInRow
+              + ", numCellsInLastRow: "
+              + numCellsInLastRow
+              + ", rowKey: "
+              + rowKey
+              + ", last5Keys: "
+              + Joiner.on(",").join(lastSeenKeys));
     }
   }
 
diff --git a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/readrows/StateMachineTest.java b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/readrows/StateMachineTest.java
new file mode 100644
index 0000000000..cbb5e7d80f
--- /dev/null
+++ b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/readrows/StateMachineTest.java
@@ -0,0 +1,77 @@
+/*
+ * Copyright 2020 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.cloud.bigtable.data.v2.stub.readrows;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.bigtable.v2.ReadRowsResponse;
+import com.google.cloud.bigtable.data.v2.models.DefaultRowAdapter;
+import com.google.cloud.bigtable.data.v2.models.Row;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.BytesValue;
+import com.google.protobuf.StringValue;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class StateMachineTest {
+  StateMachine<Row> stateMachine;
+
+  @Before
+  public void setUp() throws Exception {
+    stateMachine = new StateMachine<>(new DefaultRowAdapter().createRowBuilder());
+  }
+
+  @Test
+  public void testErrorHandlingStats() {
+    StateMachine.InvalidInputException actualError = null;
+
+    ReadRowsResponse.CellChunk chunk =
+        ReadRowsResponse.CellChunk.newBuilder()
+            .setRowKey(ByteString.copyFromUtf8("my-key1"))
+            .setFamilyName(StringValue.newBuilder().setValue("my-family"))
+            .setQualifier(BytesValue.newBuilder().setValue(ByteString.copyFromUtf8("q")))
+            .setTimestampMicros(1_000)
+            .setValue(ByteString.copyFromUtf8("my-value"))
+            .setCommitRow(true)
+            .build();
+    try {
+      stateMachine.handleChunk(chunk);
+      stateMachine.consumeRow();
+
+      stateMachine.handleChunk(
+          chunk.toBuilder().setRowKey(ByteString.copyFromUtf8("my-key2")).build());
+      stateMachine.consumeRow();
+
+      stateMachine.handleChunk(
+          chunk
+              .toBuilder()
+              .setRowKey(ByteString.copyFromUtf8("my-key3"))
+              .setValueSize(123) // invalid value size
+              .build());
+    } catch (StateMachine.InvalidInputException e) {
+      actualError = e;
+    }
+
+    assertThat(actualError).hasMessageThat().containsMatch("last5Keys: .*my-key1.*my-key2");
+    assertThat(actualError).hasMessageThat().contains("numScannedNotifications: 0");
+    assertThat(actualError).hasMessageThat().contains("numChunksProcessed: 3");
+    assertThat(actualError).hasMessageThat().contains("numCellsInRow: 0");
+    assertThat(actualError).hasMessageThat().contains("numCellsInLastRow: 1");
+  }
+}