GH-10671: Fix writingLatch lifecycle in TcpNioConnection

Inside `TcpNioConnection.ChannelInputStream.write()`, the implementation currently performs:
```
TcpNioConnection.this.writingLatch = new CountDownLatch(1);
```
This line replaces the latch with a new one after every write, even when the assembler is already waiting on the previous latch.

This leads to a lost‑notification race:

*    Assembler enters `convert()`, captures latch A, and prepares to wait.
*    Before the assembler actually calls `await()`,  the `ChannelInputStream.write()` executes and replaces latch A with B.
*    NIO reader signals B, but assembler is still waiting on A.
*    A is never signaled → assembler waits the full 60s → Timed out waiting for IO.
*    While assembler is blocked, the pipe cannot be drained → queue fills → Timed out waiting for buffer space.

Fix a lost-notify race in `TcpNioConnection` where `writingLatch` could be replaced from
`ChannelInputStream.write()`, causing the assembler to wait on a stale latch under concurrency.

Changes:

    Do not replace `writingLatch` from the pipe write path
    Clear `writingLatch` at the end of the `doRead()` cycle
    Add a regression test proving the latch is not replaced during pipe write and is cleared after read

Signed-off-by: Outman Ouharri <ouharri.outman@gmail.com>

(cherry picked from commit 18c72d1)

Author: ouharri

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java

@@ -483,6 +483,7 @@ private void doRead() throws IOException {
 		finally {
 			this.writingToPipe = false;
 			writingLatchToUse.countDown();
+			this.writingLatch = null;
 		}
 	}
 
@@ -828,7 +829,6 @@ public void write(ByteBuffer byteBuffer) throws IOException {
 					Thread.currentThread().interrupt();
 					throw new IOException("Interrupted while waiting for buffer space", e);
 				}
-				TcpNioConnection.this.writingLatch = new CountDownLatch(1);
 			}
 		}
 

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java

@@ -798,6 +798,62 @@ public void testNoMultiAccept() throws InterruptedException, IOException {
 		testMulti(false);
 	}
 
+	@Test
+	public void testWritingLatchClearedAfterRead() throws Exception {
+		SocketChannel channel = mock();
+		Socket socket = mock();
+		when(channel.socket()).thenReturn(socket);
+		when(socket.getReceiveBufferSize()).thenReturn(1024);
+
+		doAnswer(invocation -> {
+			ByteBuffer buffer = invocation.getArgument(0);
+			buffer.put("foo".getBytes());
+			return 3;
+		}).when(channel).read(Mockito.any(ByteBuffer.class));
+
+		TcpNioConnection connection = new TcpNioConnection(channel, false, false, this.nullPublisher, null);
+
+		CompositeExecutor compositeExec = compositeExecutor();
+		connection.setTaskExecutor(compositeExec);
+
+		DirectFieldAccessor dfa = new DirectFieldAccessor(connection);
+
+		ChannelInputStream originalStream =
+				TestUtils.getPropertyValue(connection, "channelInputStream", TcpNioConnection.ChannelInputStream.class);
+		assertThat(originalStream).isNotNull();
+
+		ChannelInputStream streamSpy = spy(originalStream);
+		dfa.setPropertyValue("channelInputStream", streamSpy);
+
+		AtomicReference<CountDownLatch> latchSeenDuringWrite = new AtomicReference<>();
+
+		doAnswer(invocation -> {
+			CountDownLatch currentLatch = (CountDownLatch) dfa.getPropertyValue("writingLatch");
+			if (currentLatch != null) {
+				latchSeenDuringWrite.compareAndSet(null, currentLatch);
+			}
+			return invocation.callRealMethod();
+		}).when(streamSpy).write(any(ByteBuffer.class));
+
+		Method doRead = TcpNioConnection.class.getDeclaredMethod("doRead");
+		doRead.setAccessible(true);
+
+		doRead.invoke(connection);
+
+		assertThat(latchSeenDuringWrite.get())
+				.as("writingLatch should be non-null while data is written to the pipe")
+				.isNotNull();
+
+		CountDownLatch writingLatchAfterRead = (CountDownLatch) dfa.getPropertyValue("writingLatch");
+		assertThat(writingLatchAfterRead)
+				.as("writingLatch must be null after a completed read cycle")
+				.isNull();
+
+		connection.close();
+
+		cleanupCompositeExecutor(compositeExec);
+	}
+
 	private static void testMulti(boolean multiAccept) throws InterruptedException, IOException {
 		CountDownLatch serverReadyLatch = new CountDownLatch(1);
 		CountDownLatch latch = new CountDownLatch(21);