Fix edge cases with UTF-8 strings in ChecksumResultSet (#13441)

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 resets
the buffer's write limit back to its full capacity for the next
iteration, instead of shrinking it to the size of the previous write.
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

@@ -330,7 +330,8 @@ 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));
+        // UTF-8 can require 4 bytes to represent, so we need at least that size buffer.
+        buffer = ByteBuffer.allocate(Math.max(4, Math.min(MAX_BUFFER_SIZE, length)));
       } else {
         buffer.clear();
       }
@@ -354,8 +355,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);
+  }
 }