Duplicate buffer for thread-safe aliasing. (#19405)

Fixes #10697: UnsafeDirectNioDecoder modifies shared buffer instance when performing aliased read.

I've included a test-ish reproducer in form of a unit test that fails fairly reliably, let me know whether you'd want this merged.

Closes #19405

COPYBARA_INTEGRATE_REVIEW=#19405 from mernst-github:mernst/aliasing 1d97209
PiperOrigin-RevId: 703588369

java/core/src/main/java/com/google/protobuf/CodedInputStream.java

@@ -1306,7 +1306,8 @@ static boolean isSupported() {
     }
 
     private UnsafeDirectNioDecoder(ByteBuffer buffer, boolean immutable) {
-      this.buffer = buffer;
+      // Duplicate to avoid non-threadsafe modifications in slice()
+      this.buffer = buffer.duplicate();
       address = UnsafeUtil.addressOffset(buffer);
       limit = address + buffer.limit();
       pos = address + buffer.position();

java/core/src/test/java/com/google/protobuf/CodedInputStreamTest.java

@@ -28,6 +28,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.function.Supplier;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
@@ -1405,6 +1406,41 @@ public void testReadByteBufferAliasing() throws Exception {
     }
   }
 
+  @Test
+  public void testByteBufferInputStreamReadBytesWithAliasConcurrently() {
+    int size = 127;
+    assertThat(CodedOutputStream.computeInt32SizeNoTag(size)).isEqualTo(1);
+    ByteBuffer input = ByteBuffer.allocateDirect(1 + size);
+    input.put(0, (byte) size);
+
+    Supplier<ByteString> embeddedBytes =
+        () -> {
+          try {
+            final CodedInputStream inputStream = CodedInputStream.newInstance(input, true);
+            inputStream.enableAliasing(true);
+            return inputStream.readBytes();
+          } catch (IOException e) {
+            throw new RuntimeException(e);
+          }
+        };
+
+    assertThat(embeddedBytes.get().size()).isEqualTo(size);
+
+    // Concurrent reader should have no impact.
+    int iterations = 100000;
+    new Thread(
+            () -> {
+              for (int i = 0; i < iterations; i++) {
+                ByteString unused = embeddedBytes.get();
+              }
+            })
+        .start();
+
+    for (int i = 0; i < iterations; i++) {
+      assertThat(embeddedBytes.get().size()).isEqualTo(size);
+    }
+  }
+
   @Test
   public void testIterableByteBufferInputStreamReadBytesWithAlias() throws Exception {
     ByteArrayOutputStream byteArrayStream = new ByteArrayOutputStream();