fix(java-spanner): Fix edge cases with UTF-8 strings in ChecksumResultSet

1. Ensure a minimum capacity of 4 bytes when allocating the buffer, this is the max size of a UTF-8 character. However, the java length representation is being used in this code for the byte buffer allocation, which may be too small for a single utf-8 character.

2. Use buffer.clear() instead of the second buffer.flip(). This is for mixed multi-byte utf-8 characters and single-byte characters. A test was added to show this passing, and it fails with flip() vs clear().

Fixes #13440

Author: Noremac201

java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/connection/ChecksumResultSet.java

@@ -329,7 +329,7 @@ private void putString(String stringValue) {
       if (buffer == null || (buffer.capacity() < MAX_BUFFER_SIZE && buffer.capacity() < length)) {
         // Create a ByteBuffer with a maximum buffer size.
         // This buffer is re-used for all string values in the result set.
-        buffer = ByteBuffer.allocate(Math.min(MAX_BUFFER_SIZE, length));
+        buffer = ByteBuffer.allocate(Math.max(4, Math.min(MAX_BUFFER_SIZE, length)));
       } else {
         buffer.clear();
       }
@@ -349,8 +349,8 @@ private void putString(String stringValue) {
         buffer.flip();
         // Put the bytes from the buffer into the digest.
         digest.update(buffer);
-        // Flip the buffer again, so we can repeat and write to the start of the buffer again.
-        buffer.flip();
+        // Clear the buffer, so we can repeat and write to the start of the buffer again.
+        buffer.clear();
       }
     }
   }

java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/ChecksumResultSetTest.java

@@ -448,4 +448,42 @@ public void testRetry() {
           () -> resultSet.retry(abortedException));
     }
   }
+
+  @Test
+  public void testEmptyString() {
+    ChecksumResultSet resultSet = createStringValChecksumResultSet("");
+    assertTrue(resultSet.next());
+  }
+
+  @Test
+  public void testSingleCharMultiByteString() {
+    ChecksumResultSet resultSet = createStringValChecksumResultSet("ä");
+    assertTrue(resultSet.next());
+  }
+
+  @Test
+  public void testLongMixedUtf8String() {
+    ChecksumResultSet resultSet = createStringValChecksumResultSet("aaa\uD841\uDF0E");
+    assertTrue(resultSet.next());
+  }
+
+  private ChecksumResultSet createStringValChecksumResultSet(String value) {
+    Type type = Type.struct(StructField.of("stringVal", Type.string()));
+    Struct row = Struct.newBuilder().set("stringVal").to(value).build();
+
+    ParsedStatement parsedStatement = mock(ParsedStatement.class);
+    Statement statement = Statement.of("select * from foo");
+    when(parsedStatement.getStatement()).thenReturn(statement);
+    ReadWriteTransaction transaction = mock(ReadWriteTransaction.class);
+    when(transaction.runWithRetry(any(Callable.class)))
+        .thenAnswer(invocationOnMock -> ((Callable<?>) invocationOnMock.getArgument(0)).call());
+    when(transaction.getStatementExecutor()).thenReturn(mock(StatementExecutor.class));
+
+    ResultSet queryResult = ResultSets.forRows(type, ImmutableList.of(row));
+    return new ChecksumResultSet(
+        transaction,
+        DirectExecuteResultSet.ofResultSet(queryResult),
+        parsedStatement,
+        AnalyzeMode.NONE);
+  }
 }