Fix header copy aliasing that breaks message delivery

src/main/java/io/nats/client/impl/Headers.java

@@ -86,6 +86,8 @@ public Headers(@Nullable Headers headers, boolean readOnly, String @Nullable []
 		// (add/put/remove/clear) already short-circuits on `readOnly` before
 		// touching the map, and the map is never exposed by reference.
 		this.valuesMap = new HashMap<>(headers.valuesMap);
+		// Keep values independent so mutations cannot invalidate another Headers' length bookkeeping.
+		this.valuesMap.replaceAll((key, values) -> new ArrayList<>(values));
 		this.lengthMap = new HashMap<>(headers.lengthMap);
 		this.dataLength = headers.dataLength;
 

src/test/java/io/nats/client/PublishTests.java

@@ -20,6 +20,8 @@
 
 import java.net.SocketException;
 import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collections;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -33,6 +35,30 @@
 import static org.junit.jupiter.api.Assertions.*;
 
 public class PublishTests {
+    @Test
+    public void testPublishWithIndependentlyModifiedHeaderCopy() throws Exception {
+        try (NatsTestServer ts = new NatsTestServer(false);
+             Connection nc = Nats.connect(ts.getURI())) {
+            Subscription sub = nc.subscribe("header.copy");
+            Headers template = new Headers().add("Trace", "base");
+            Headers extended = new Headers(template).add("Trace", "derived");
+            byte[] body = "payload".getBytes(StandardCharsets.UTF_8);
+
+            nc.publish("header.copy", template, body);
+            nc.publish("header.copy", extended, body);
+
+            Message first = sub.nextMessage(2000);
+            assertNotNull(first, "Publishing the original headers must still deliver the message");
+            assertEquals(Collections.singletonList("base"), first.getHeaders().get("Trace"));
+            assertArrayEquals(body, first.getData());
+
+            Message second = sub.nextMessage(2000);
+            assertNotNull(second, "Publishing the extended copy must deliver the message");
+            assertEquals(Arrays.asList("base", "derived"), second.getHeaders().get("Trace"));
+            assertArrayEquals(body, second.getData());
+        }
+    }
+
     @Test
     public void throwsIfClosedOnPublish() {
         assertThrows(IllegalStateException.class, () -> {

src/test/java/io/nats/client/impl/HeadersTests.java

@@ -2,6 +2,8 @@
 
 import io.nats.client.support.*;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 import java.nio.charset.StandardCharsets;
 import java.util.*;
@@ -26,6 +28,32 @@ public class HeadersTests {
     private static final String VAL6 = "val6";
     private static final String EMPTY = "";
 
+    @Test
+    public void testAddingToCopyDoesNotChangeOriginal() {
+        Headers original = new Headers().add(KEY1, VAL1);
+        Headers copy = new Headers(original);
+        copy.add(KEY1, VAL2);
+
+        assertEquals(Collections.singletonList(VAL1), original.get(KEY1));
+        assertEquals(Arrays.asList(VAL1, VAL2), copy.get(KEY1));
+        validateDirtyAndLength(original);
+        validateDirtyAndLength(copy);
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    public void testAddingToOriginalDoesNotChangeCopy(boolean readOnly) {
+        Headers original = new Headers().add(KEY1, VAL1).add(KEY2, VAL2);
+        Headers copy = new Headers(original, readOnly, new String[] {KEY2});
+        original.add(KEY1, VAL3);
+
+        assertEquals(Collections.singletonList(VAL1), copy.get(KEY1));
+        assertFalse(copy.containsKey(KEY2));
+        assertEquals(Arrays.asList(VAL1, VAL3), original.get(KEY1));
+        validateDirtyAndLength(original);
+        validateDirtyAndLength(copy);
+    }
+
     @Test
     public void add_key_strings_works() {
         add(