This is an automated email from the ASF dual-hosted git repository.
vavrtom pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/qpid-broker-j.git
The following commit(s) were added to refs/heads/main by this push:
new d65a81b813 QPID-8749: [Broker-J] AMQP connections may not close while
processing pending writes (#423)
d65a81b813 is described below
commit d65a81b8130b729baaa1cd9a3f2bb09e3cdc5447
Author: Daniil Kirilyuk <[email protected]>
AuthorDate: Fri Jul 31 13:44:52 2026 +0200
QPID-8749: [Broker-J] AMQP connections may not close while processing
pending writes (#423)
---
.../qpid/server/bytebuffer/QpidByteBuffer.java | 7 +
.../server/bytebuffer/QpidByteBufferFactory.java | 46 +++-
.../apache/qpid/server/model/port/AmqpPort.java | 8 +
.../qpid/server/model/port/AmqpPortImpl.java | 32 ++-
.../server/transport/NonBlockingConnection.java | 47 ++--
.../NonBlockingConnectionPlainDelegate.java | 28 +--
.../qpid/server/bytebuffer/QpidByteBufferTest.java | 109 +++++++++
.../qpid/server/model/port/AmqpPortImplTest.java | 59 +++++
.../NonBlockingConnectionIdleTimeoutTest.java | 249 +++++++++++++++++++++
.../NonBlockingConnectionPlainDelegateTest.java | 112 +++++++++
.../transport/NonBlockingConnectionTest.java | 1 +
.../server/transport/TCPandSSLTransportTest.java | 1 +
12 files changed, 654 insertions(+), 45 deletions(-)
diff --git
a/broker-core/src/main/java/org/apache/qpid/server/bytebuffer/QpidByteBuffer.java
b/broker-core/src/main/java/org/apache/qpid/server/bytebuffer/QpidByteBuffer.java
index eb49b56f2e..92965f248f 100644
---
a/broker-core/src/main/java/org/apache/qpid/server/bytebuffer/QpidByteBuffer.java
+++
b/broker-core/src/main/java/org/apache/qpid/server/bytebuffer/QpidByteBuffer.java
@@ -84,6 +84,13 @@ public interface QpidByteBuffer extends AutoCloseable
return QpidByteBufferFactory.write(channel, qpidByteBuffers);
}
+ static long write(final GatheringByteChannel channel,
+ final Collection<QpidByteBuffer> qpidByteBuffers,
+ final int maxBuffers) throws IOException
+ {
+ return QpidByteBufferFactory.write(channel, qpidByteBuffers,
maxBuffers);
+ }
+
static QpidByteBuffer wrap(ByteBuffer wrap)
{
return QpidByteBufferFactory.wrap(wrap);
diff --git
a/broker-core/src/main/java/org/apache/qpid/server/bytebuffer/QpidByteBufferFactory.java
b/broker-core/src/main/java/org/apache/qpid/server/bytebuffer/QpidByteBufferFactory.java
index fd74860abf..e4fc3a84d7 100644
---
a/broker-core/src/main/java/org/apache/qpid/server/bytebuffer/QpidByteBufferFactory.java
+++
b/broker-core/src/main/java/org/apache/qpid/server/bytebuffer/QpidByteBufferFactory.java
@@ -265,15 +265,49 @@ final class QpidByteBufferFactory
}
}
- static long write(GatheringByteChannel channel, Collection<QpidByteBuffer>
qpidByteBuffers)
- throws IOException
+ static long write(final GatheringByteChannel channel, final
Collection<QpidByteBuffer> qpidByteBuffers) throws IOException
{
- List<ByteBuffer> byteBuffers = new ArrayList<>();
- for (QpidByteBuffer qpidByteBuffer : qpidByteBuffers)
+ return write(channel, qpidByteBuffers, Integer.MAX_VALUE);
+ }
+
+ static long write(final GatheringByteChannel channel,
+ final Collection<QpidByteBuffer> qpidByteBuffers,
+ final int maxBuffers) throws IOException
+ {
+ if (maxBuffers <= 0)
{
- Collections.addAll(byteBuffers,
getUnderlyingBuffers(qpidByteBuffer));
+ throw new IllegalArgumentException("Maximum number of buffers must
be greater than zero");
}
- return channel.write(byteBuffers.toArray(new
ByteBuffer[byteBuffers.size()]));
+
+ final List<ByteBuffer> byteBuffers = new ArrayList<>();
+
+ outer: for (final QpidByteBuffer qpidByteBuffer : qpidByteBuffers)
+ {
+ if (qpidByteBuffer instanceof MultiQpidByteBuffer
multiQpidByteBuffer)
+ {
+ for (final SingleQpidByteBuffer fragment :
multiQpidByteBuffer.getFragments())
+ {
+ if (fragment.hasRemaining())
+ {
+ byteBuffers.add(fragment.getUnderlyingBuffer());
+ if (byteBuffers.size() == maxBuffers)
+ {
+ break outer;
+ }
+ }
+ }
+ }
+ else if (qpidByteBuffer.hasRemaining())
+ {
+ byteBuffers.add(((SingleQpidByteBuffer)
qpidByteBuffer).getUnderlyingBuffer());
+ if (byteBuffers.size() == maxBuffers)
+ {
+ break;
+ }
+ }
+ }
+
+ return byteBuffers.isEmpty() ? 0L :
channel.write(byteBuffers.toArray(new ByteBuffer[0]));
}
static QpidByteBuffer wrap(ByteBuffer wrap)
diff --git
a/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPort.java
b/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPort.java
index 229a6ee7f0..2d9a9badd4 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPort.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPort.java
@@ -159,6 +159,14 @@ public interface AmqpPort<X extends AmqpPort<X>> extends
Port<X>
" If the connection does not close this time, it will be aborted.")
long DEFAULT_FINAL_WRITE_TIMEOUT = 1000L;
+ String MAX_GATHERING_WRITE_BUFFERS =
"qpid.port.amqp.maxGatheringWriteBuffers";
+ int MINIMUM_MAX_GATHERING_WRITE_BUFFERS = 1;
+ int MAXIMUM_MAX_GATHERING_WRITE_BUFFERS = 1024;
+ @SuppressWarnings("unused")
+ @ManagedContextDefault(name = MAX_GATHERING_WRITE_BUFFERS,
+ description = "Maximum number of underlying NIO buffers included
in one gathering socket write.")
+ int DEFAULT_MAX_GATHERING_WRITE_BUFFERS =
MAXIMUM_MAX_GATHERING_WRITE_BUFFERS;
+
@ManagedAttribute( defaultValue = AmqpPort.DEFAULT_AMQP_TCP_NO_DELAY )
boolean isTcpNoDelay();
diff --git
a/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPortImpl.java
b/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPortImpl.java
index ab810b603a..07ab4d181c 100644
---
a/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPortImpl.java
+++
b/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPortImpl.java
@@ -350,20 +350,37 @@ public class AmqpPortImpl extends
AbstractPort<AmqpPortImpl> implements AmqpPort
}
validateThreadPoolSettings(this);
+ validateMaxGatheringWriteBuffers(getContextValue(Integer.class,
MAX_GATHERING_WRITE_BUFFERS));
}
@Override
protected void validateChange(final ConfiguredObject<?>
proxyForValidation, final Set<String> changedAttributes)
{
super.validateChange(proxyForValidation, changedAttributes);
- AmqpPort changed = (AmqpPort) proxyForValidation;
+ final AmqpPort<?> changed = (AmqpPort<?>) proxyForValidation;
if (changedAttributes.contains(THREAD_POOL_SIZE) ||
changedAttributes.contains(NUMBER_OF_SELECTORS))
{
validateThreadPoolSettings(changed);
}
+ if (changedAttributes.contains(CONTEXT))
+ {
+ final String configuredValue =
changed.getContext().get(MAX_GATHERING_WRITE_BUFFERS);
+ try
+ {
+ validateMaxGatheringWriteBuffers(configuredValue == null
+ ? DEFAULT_MAX_GATHERING_WRITE_BUFFERS
+ : Integer.parseInt(configuredValue));
+ }
+ catch (NumberFormatException e)
+ {
+ final String message = "Maximum number of gathering write
buffers on Port %s must be an integer"
+ .formatted(getName());
+ throw new IllegalConfigurationException(message, e);
+ }
+ }
}
- private void validateThreadPoolSettings(final AmqpPort changed)
+ private void validateThreadPoolSettings(final AmqpPort<?> changed)
{
if (changed.getThreadPoolSize() < 1)
{
@@ -379,6 +396,17 @@ public class AmqpPortImpl extends
AbstractPort<AmqpPortImpl> implements AmqpPort
}
}
+ private void validateMaxGatheringWriteBuffers(final int
maxGatheringWriteBuffers)
+ {
+ if (maxGatheringWriteBuffers < MINIMUM_MAX_GATHERING_WRITE_BUFFERS ||
maxGatheringWriteBuffers > MAXIMUM_MAX_GATHERING_WRITE_BUFFERS)
+ {
+ final String message = "Maximum number of gathering write buffers
%d on Port %s must be between %d and %d"
+ .formatted(maxGatheringWriteBuffers, getName(),
MINIMUM_MAX_GATHERING_WRITE_BUFFERS,
+ MAXIMUM_MAX_GATHERING_WRITE_BUFFERS);
+ throw new IllegalConfigurationException(message);
+ }
+ }
+
private SSLContext createSslContext()
{
KeyStore keyStore = getKeyStore();
diff --git
a/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnection.java
b/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnection.java
index d8a8857d82..343e2b3343 100644
---
a/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnection.java
+++
b/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnection.java
@@ -63,6 +63,7 @@ public class NonBlockingConnection implements
ServerNetworkConnection, ByteBuffe
private final Runnable _onTransportEncryptionAction;
private final int _finalWriteThreshold;
private final long _finalWriteTimeout;
+ private final int _maxGatheringWriteBuffers;
private volatile boolean _fullyWritten = true;
@@ -83,8 +84,8 @@ public class NonBlockingConnection implements
ServerNetworkConnection, ByteBuffe
private volatile long _bufferedSize;
private String _selectedHost;
- public NonBlockingConnection(SocketChannel socketChannel,
- ProtocolEngine protocolEngine,
+ public NonBlockingConnection(final SocketChannel socketChannel,
+ final ProtocolEngine protocolEngine,
final Set<TransportEncryption> encryptionSet,
final Runnable onTransportEncryptionAction,
final NetworkConnectionScheduler scheduler,
@@ -118,6 +119,7 @@ public class NonBlockingConnection implements
ServerNetworkConnection, ByteBuffe
}
_finalWriteThreshold = port.getContextValue(Integer.class,
AmqpPort.FINAL_WRITE_THRESHOLD);
_finalWriteTimeout = port.getContextValue(Long.class,
AmqpPort.FINAL_WRITE_TIMEOUT);
+ _maxGatheringWriteBuffers = port.getContextValue(Integer.class,
AmqpPort.MAX_GATHERING_WRITE_BUFFERS);
final Broker broker = (Broker<?>) port.getParent();
_eventLogger = broker.getEventLogger();
@@ -259,33 +261,36 @@ public class NonBlockingConnection implements
ServerNetworkConnection, ByteBuffe
{
try
{
- long currentTime = System.currentTimeMillis();
- int tick = getTicker().getTimeToNextTick(currentTime);
+ final long currentTime = System.currentTimeMillis();
+ final int tick = getTicker().getTimeToNextTick(currentTime);
if (tick <= 0)
{
getTicker().tick(currentTime);
}
+ _scheduledTime = 0L;
- _protocolEngine.setIOThread(Thread.currentThread());
-
- boolean processPendingComplete = processPending();
-
- if(processPendingComplete)
+ if (!_closed.get())
{
- _pendingIterator = null;
- _protocolEngine.setTransportBlockedForWriting(false);
- boolean dataRead = doRead();
- _protocolEngine.setTransportBlockedForWriting(!doWrite());
+ _protocolEngine.setIOThread(Thread.currentThread());
+
+ final boolean processPendingComplete = processPending();
- if (!_fullyWritten || dataRead || (_delegate.needsWork()
&& _delegate.getNetInputBuffer().position() != 0))
+ if (processPendingComplete)
+ {
+ _pendingIterator = null;
+ _protocolEngine.setTransportBlockedForWriting(false);
+ final boolean dataRead = doRead();
+
_protocolEngine.setTransportBlockedForWriting(!doWrite());
+
+ if (!_fullyWritten || dataRead ||
(_delegate.needsWork() && _delegate.getNetInputBuffer().position() != 0))
+ {
+ _protocolEngine.notifyWork();
+ }
+ }
+ else
{
_protocolEngine.notifyWork();
}
-
- }
- else
- {
- _protocolEngine.notifyWork();
}
}
@@ -523,9 +528,9 @@ public class NonBlockingConnection implements
ServerNetworkConnection, ByteBuffe
}
}
- long writeToTransport(Collection<QpidByteBuffer> buffers) throws
IOException
+ long writeToTransport(final Collection<QpidByteBuffer> buffers) throws
IOException
{
- long written = QpidByteBuffer.write(_socketChannel, buffers);
+ final long written = QpidByteBuffer.write(_socketChannel, buffers,
_maxGatheringWriteBuffers);
if (LOGGER.isDebugEnabled())
{
LOGGER.debug("Written " + written + " bytes");
diff --git
a/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionPlainDelegate.java
b/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionPlainDelegate.java
index d355f4b613..5e1f398688 100644
---
a/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionPlainDelegate.java
+++
b/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionPlainDelegate.java
@@ -99,29 +99,25 @@ public class NonBlockingConnectionPlainDelegate implements
NonBlockingConnection
}
-
@Override
- public WriteResult doWrite(Collection<QpidByteBuffer> buffers) throws
IOException
+ public WriteResult doWrite(final Collection<QpidByteBuffer> buffers)
throws IOException
{
- long bytesToWrite = 0L;
- if(!buffers.isEmpty())
- {
- for (QpidByteBuffer buf : buffers)
- {
- bytesToWrite += buf.remaining();
- }
- }
- if(bytesToWrite == 0L)
+ if (buffers.isEmpty())
{
return new WriteResult(true, 0);
}
- else
- {
- long bytesWritten = _parent.writeToTransport(buffers);
- return new WriteResult(bytesWritten >= bytesToWrite, bytesWritten);
+ final long bytesWritten = _parent.writeToTransport(buffers);
+ boolean complete = true;
+ for (final QpidByteBuffer buffer : buffers)
+ {
+ if (buffer.hasRemaining())
+ {
+ complete = false;
+ break;
+ }
}
-
+ return new WriteResult(complete, bytesWritten);
}
@Override
diff --git
a/broker-core/src/test/java/org/apache/qpid/server/bytebuffer/QpidByteBufferTest.java
b/broker-core/src/test/java/org/apache/qpid/server/bytebuffer/QpidByteBufferTest.java
index a3e4523ff5..6b9ad47ca6 100644
---
a/broker-core/src/test/java/org/apache/qpid/server/bytebuffer/QpidByteBufferTest.java
+++
b/broker-core/src/test/java/org/apache/qpid/server/bytebuffer/QpidByteBufferTest.java
@@ -27,6 +27,9 @@ import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
@@ -38,8 +41,11 @@ import java.nio.BufferOverflowException;
import java.nio.BufferUnderflowException;
import java.nio.ByteBuffer;
import java.nio.InvalidMarkException;
+import java.nio.channels.GatheringByteChannel;
import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.List;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@@ -922,6 +928,109 @@ public class QpidByteBufferTest extends UnitTestBase
}
}
+ @Test
+ public void testGatheringWriteLimitsUnderlyingBuffers() throws Exception
+ {
+ final List<QpidByteBuffer> buffers = new ArrayList<>();
+ final List<Integer> batchSizes = new ArrayList<>();
+
+ try (final ByteArrayOutputStream output = new ByteArrayOutputStream();
+ final GatheringByteChannel channel = createWritingChannel(output,
batchSizes))
+ {
+ for (int i = 0; i < 5; i++)
+ {
+ buffers.add(QpidByteBuffer.wrap(new byte[] {(byte) i}));
+ }
+
+ while (hasRemaining(buffers))
+ {
+ QpidByteBuffer.write(channel, buffers, 2);
+ }
+
+ assertEquals(List.of(2, 2, 1), batchSizes, "Unexpected gathering
write batch sizes");
+ assertArrayEquals(new byte[] {0, 1, 2, 3, 4},
output.toByteArray(), "Gathering write changed buffer order");
+ }
+ finally
+ {
+ for (final QpidByteBuffer buffer : buffers)
+ {
+ buffer.dispose();
+ }
+ }
+ }
+
+ @Test
+ public void testGatheringWriteLimitsCompositeBufferFragments() throws
Exception
+ {
+ final byte[] content = new byte[BUFFER_FRAGMENT_SIZE * 3];
+ for (int i = 0; i < content.length; i++)
+ {
+ content[i] = (byte) i;
+ }
+
+ final List<Integer> batchSizes = new ArrayList<>();
+
+ try (final ByteArrayOutputStream output = new ByteArrayOutputStream();
+ final GatheringByteChannel channel = createWritingChannel(output,
batchSizes);
+ final QpidByteBuffer buffer =
QpidByteBuffer.allocateDirect(content.length))
+ {
+ buffer.put(content);
+ buffer.flip();
+
+ assertEquals(BUFFER_FRAGMENT_SIZE * 2,
QpidByteBuffer.write(channel, List.of(buffer), 2),
+ "Unexpected first gathering write size");
+ assertEquals(BUFFER_FRAGMENT_SIZE, QpidByteBuffer.write(channel,
List.of(buffer), 2),
+ "Unexpected second gathering write size");
+
+ assertEquals(List.of(2, 1), batchSizes, "Exhausted fragments
should not count towards the limit");
+ assertArrayEquals(content, output.toByteArray(), "Gathering write
changed fragment order");
+ }
+ }
+
+ @Test
+ public void testGatheringWriteRejectsInvalidMaximum() throws Exception
+ {
+ try (final GatheringByteChannel channel =
mock(GatheringByteChannel.class))
+ {
+ assertThrows(IllegalArgumentException.class, () ->
QpidByteBuffer.write(channel, List.of(_parent), 0));
+ assertThrows(IllegalArgumentException.class, () ->
QpidByteBuffer.write(channel, List.of(_parent), -1));
+ }
+ }
+
+ private GatheringByteChannel createWritingChannel(final
ByteArrayOutputStream output,
+ final List<Integer>
batchSizes) throws IOException
+ {
+ final GatheringByteChannel channel = mock(GatheringByteChannel.class);
+ when(channel.write(any(ByteBuffer[].class))).thenAnswer(invocation ->
+ {
+ final ByteBuffer[] byteBuffers = invocation.getArgument(0);
+ batchSizes.add(byteBuffers.length);
+ long written = 0;
+ for (final ByteBuffer byteBuffer : byteBuffers)
+ {
+ while (byteBuffer.hasRemaining())
+ {
+ output.write(byteBuffer.get());
+ written++;
+ }
+ }
+ return written;
+ });
+ return channel;
+ }
+
+ private boolean hasRemaining(final List<QpidByteBuffer> buffers)
+ {
+ for (final QpidByteBuffer buffer : buffers)
+ {
+ if (buffer.hasRemaining())
+ {
+ return true;
+ }
+ }
+ return false;
+ }
+
private void doDeflateInflate(final byte[] input,
final QpidByteBuffer inputBuf,
final boolean direct) throws IOException
diff --git
a/broker-core/src/test/java/org/apache/qpid/server/model/port/AmqpPortImplTest.java
b/broker-core/src/test/java/org/apache/qpid/server/model/port/AmqpPortImplTest.java
index 5da1320d46..45eaaec0da 100644
---
a/broker-core/src/test/java/org/apache/qpid/server/model/port/AmqpPortImplTest.java
+++
b/broker-core/src/test/java/org/apache/qpid/server/model/port/AmqpPortImplTest.java
@@ -42,6 +42,8 @@ import java.util.Set;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import org.apache.qpid.server.configuration.IllegalConfigurationException;
import org.apache.qpid.server.configuration.updater.CurrentThreadTaskExecutor;
@@ -234,6 +236,63 @@ public class AmqpPortImplTest extends UnitTestBase
"Exception not thrown for number of selectors equal to thread
pool size");
}
+ @Test
+ public void testDefaultMaxGatheringWriteBuffers()
+ {
+ _port = createPort(getTestName());
+
+ assertEquals(AmqpPort.DEFAULT_MAX_GATHERING_WRITE_BUFFERS,
+ _port.getContextValue(Integer.class,
AmqpPort.MAX_GATHERING_WRITE_BUFFERS),
+ "Unexpected default maximum gathering write buffer count");
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = { 1, 16, 1024 })
+ public void testValidMaxGatheringWriteBuffersOnCreate(final int value)
+ {
+ _port = createPort(getTestName(), Map.of(AmqpPort.CONTEXT,
Map.of(AmqpPort.MAX_GATHERING_WRITE_BUFFERS,
+ String.valueOf(value))));
+
+ assertEquals(value, _port.getContextValue(Integer.class,
AmqpPort.MAX_GATHERING_WRITE_BUFFERS),
+ "Unexpected maximum gathering write buffer count");
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = { -1, 0, 1025 })
+ public void testInvalidMaxGatheringWriteBuffersOnCreate(final int value)
+ {
+ final Map<String, Object> attributes = Map.of(AmqpPort.CONTEXT,
Map.of(AmqpPort.MAX_GATHERING_WRITE_BUFFERS,
+ String.valueOf(value)));
+
+ assertThrows(IllegalConfigurationException.class, () ->
createPort(getTestName(), attributes),
+ "Creation should reject an out-of-range maximum gathering
write buffer count");
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = { -1, 0, 1025 })
+ public void testInvalidMaxGatheringWriteBuffersOnChange(final int value)
+ {
+ _port = createPort(getTestName());
+ final Map<String, Object> context =
Map.of(AmqpPort.MAX_GATHERING_WRITE_BUFFERS, String.valueOf(value));
+
+ assertThrows(IllegalConfigurationException.class, () ->
+ _port.setAttributes(Map.of(AmqpPort.CONTEXT, context)),
+ "Update should reject an out-of-range maximum gathering write
buffer count");
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = { 1, 16, 1024 })
+ public void testValidMaxGatheringWriteBuffersOnChange(final int value)
+ {
+ _port = createPort(getTestName());
+ final Map<String, Object> context =
Map.of(AmqpPort.MAX_GATHERING_WRITE_BUFFERS, String.valueOf(value));
+
+ _port.setAttributes(Map.of(AmqpPort.CONTEXT, context));
+
+ assertEquals(value, _port.getContextValue(Integer.class,
AmqpPort.MAX_GATHERING_WRITE_BUFFERS),
+ "Unexpected updated maximum gathering write buffer count");
+ }
+
@Test
public void testConnectionCounting()
{
diff --git
a/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionIdleTimeoutTest.java
b/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionIdleTimeoutTest.java
new file mode 100644
index 0000000000..2d344a2ba5
--- /dev/null
+++
b/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionIdleTimeoutTest.java
@@ -0,0 +1,249 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.qpid.server.transport;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.lang.reflect.Field;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.net.SocketAddress;
+import java.nio.channels.SocketChannel;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.logging.EventLogger;
+import org.apache.qpid.server.model.Broker;
+import org.apache.qpid.server.model.port.AmqpPort;
+import org.apache.qpid.server.transport.network.TransportEncryption;
+import org.apache.qpid.test.utils.UnitTestBase;
+
+class NonBlockingConnectionIdleTimeoutTest extends UnitTestBase
+{
+ private ProtocolEngine _protocolEngine;
+ private AggregateTicker _ticker;
+ private NonBlockingConnection _connection;
+
+ @BeforeEach
+ void beforeEach()
+ {
+ final SocketAddress localAddress = mock(SocketAddress.class);
+ when(localAddress.toString()).thenReturn("127.0.0.1:5672");
+ final Socket socket = mock(Socket.class);
+ when(socket.getRemoteSocketAddress()).thenReturn(new
InetSocketAddress("localhost", 1000));
+ when(socket.getLocalSocketAddress()).thenReturn(localAddress);
+ final SocketChannel socketChannel = mock(SocketChannel.class);
+ when(socketChannel.socket()).thenReturn(socket);
+
+ _ticker = mock(AggregateTicker.class);
+ _protocolEngine = mock(ProtocolEngine.class);
+ when(_protocolEngine.getAggregateTicker()).thenReturn(_ticker);
+ when(_protocolEngine.processPendingIterator()).thenAnswer(invocation
-> Collections.emptyIterator());
+
+ final NetworkConnectionScheduler scheduler =
mock(NetworkConnectionScheduler.class);
+ final AmqpPort<?> port = mock(AmqpPort.class);
+ when(port.getNetworkBufferSize()).thenReturn(1024);
+ when(port.getContextValue(Integer.class,
AmqpPort.FINAL_WRITE_THRESHOLD)).thenReturn(100);
+ when(port.getContextValue(Long.class,
AmqpPort.FINAL_WRITE_TIMEOUT)).thenReturn(100L);
+ when(port.getContextValue(Integer.class,
AmqpPort.MAX_GATHERING_WRITE_BUFFERS)).thenReturn(1024);
+ final EventLogger eventLogger = mock(EventLogger.class);
+ final Broker<?> broker = mock(Broker.class);
+ when(broker.getEventLogger()).thenReturn(eventLogger);
+ doReturn(broker).when(port).getParent();
+
+ _connection = new NonBlockingConnection(socketChannel,
_protocolEngine, Set.of(TransportEncryption.NONE),
+ () -> { }, scheduler, port);
+ final SelectorThread.SelectionTask _selectionTask =
mock(SelectorThread.SelectionTask.class);
+ _connection.setSelectionTask(_selectionTask);
+ }
+
+ @AfterEach
+ void afterEach()
+ {
+ _connection.close();
+ _connection.doWork();
+ }
+
+ @Test
+ void immediateRerunUsesCurrentTimeForTicker()
+ {
+ final List<Long> observedScheduledTimes = new ArrayList<>();
+ when(_ticker.getTimeToNextTick(anyLong())).thenAnswer(invocation ->
+ {
+ final long scheduledTime = _connection.getScheduledTime();
+ observedScheduledTimes.add(scheduledTime);
+ return scheduledTime > 0 ? 1 : 0;
+ });
+ when(_ticker.tick(anyLong())).thenReturn(Integer.MAX_VALUE);
+
+ assertTrue(_connection.setScheduled(), "Connection should be
scheduled");
+ final long initialScheduledTime = _connection.getScheduledTime();
+
+ _connection.doWork();
+
+ assertEquals(initialScheduledTime, observedScheduledTimes.get(0),
+ "First ticker evaluation should see the original scheduled
time");
+ assertEquals(0, _connection.getScheduledTime(),
+ "Scheduled time should be cleared after the first ticker
evaluation");
+ verify(_ticker, never()).tick(anyLong());
+
+ _connection.doWork();
+
+ assertEquals(0, observedScheduledTimes.get(1), "Immediate rerun should
evaluate the ticker against current time");
+ verify(_ticker, times(1)).tick(anyLong());
+ }
+
+ @Test
+ void tickerCloseSkipsOrdinaryIoAndRunsShutdown() throws Exception
+ {
+ final NonBlockingConnectionDelegate delegate =
mock(NonBlockingConnectionDelegate.class);
+ when(delegate.doWrite(any())).thenReturn(new
NonBlockingConnectionDelegate.WriteResult(true, 0));
+ injectDelegate(delegate);
+ when(_ticker.getTimeToNextTick(anyLong())).thenReturn(0);
+ doAnswer(invocation ->
+ {
+ _connection.close();
+ return Integer.MAX_VALUE;
+ }).when(_ticker).tick(anyLong());
+ assertTrue(_connection.setScheduled(), "Connection should be
scheduled");
+
+ _connection.doWork();
+
+ verify(_protocolEngine, never()).processPendingIterator();
+ verify(_protocolEngine,
never()).setTransportBlockedForWriting(anyBoolean());
+ verify(delegate, never()).readyForRead();
+ verify(delegate, times(1)).doWrite(any());
+ verify(_protocolEngine, times(1)).closed();
+ verify(delegate, times(1)).shutdownInput();
+ verify(delegate, times(1)).shutdownOutput();
+ }
+
+ @Test
+ void blockedPendingOutputDoesNotStarveReadIdleTicker() throws Exception
+ {
+ final int readDelay = 1000;
+ final long lastReadTime = System.currentTimeMillis() - 2 * readDelay;
+ final long timeoutTime = lastReadTime + readDelay;
+ final AtomicBoolean readerIdle = new AtomicBoolean();
+ final AggregateTicker aggregateTicker = new AggregateTicker();
+ when(_protocolEngine.getAggregateTicker()).thenReturn(aggregateTicker);
+ when(_protocolEngine.getLastReadTime()).thenReturn(lastReadTime);
+ doAnswer(invocation ->
+ {
+ readerIdle.set(true);
+ _connection.close();
+ return null;
+ }).when(_protocolEngine).readerIdle();
+ aggregateTicker.addTicker(new ServerIdleReadTimeoutTicker(_connection,
_protocolEngine, readDelay));
+
+ final Iterator<Runnable> pendingIterator = new Iterator<>()
+ {
+ private boolean _outputQueued;
+
+ @Override
+ public boolean hasNext()
+ {
+ return true;
+ }
+
+ @Override
+ public Runnable next()
+ {
+ if (!_outputQueued)
+ {
+ _outputQueued = true;
+ return () -> _connection.send(QpidByteBuffer.wrap(new
byte[1024]));
+ }
+ return () -> { };
+ }
+ };
+
when(_protocolEngine.processPendingIterator()).thenReturn(pendingIterator);
+
+ final NonBlockingConnectionDelegate delegate =
mock(NonBlockingConnectionDelegate.class);
+ when(delegate.doWrite(any())).thenAnswer(invocation ->
+ {
+ if (!readerIdle.get())
+ {
+ return new NonBlockingConnectionDelegate.WriteResult(false, 0);
+ }
+
+ long bytesConsumed = 0;
+ final Collection<QpidByteBuffer> buffers =
invocation.getArgument(0);
+ for (final QpidByteBuffer buffer : buffers)
+ {
+ bytesConsumed += buffer.remaining();
+ buffer.position(buffer.limit());
+ }
+ return new NonBlockingConnectionDelegate.WriteResult(true,
bytesConsumed);
+ });
+ injectDelegate(delegate);
+
+ assertTrue(_connection.setScheduled(), "Connection should be
scheduled");
+ injectScheduledTime(timeoutTime - 1);
+
+ _connection.doWork();
+
+ assertEquals(0, _connection.getScheduledTime(), "Scheduled time should
be cleared while pending output remains blocked");
+ verify(_protocolEngine, never()).readerIdle();
+ verify(delegate, times(2)).doWrite(any());
+
+ _connection.doWork();
+
+ verify(_protocolEngine, times(1)).readerIdle();
+ verify(_protocolEngine, times(1)).processPendingIterator();
+ verify(delegate, times(3)).doWrite(any());
+ verify(_protocolEngine, times(1)).closed();
+ }
+
+ private void injectDelegate(final NonBlockingConnectionDelegate delegate)
throws Exception
+ {
+ final Field delegateField =
NonBlockingConnection.class.getDeclaredField("_delegate");
+ delegateField.setAccessible(true);
+ delegateField.set(_connection, delegate);
+ }
+
+ private void injectScheduledTime(final long scheduledTime) throws Exception
+ {
+ final Field scheduledTimeField =
NonBlockingConnection.class.getDeclaredField("_scheduledTime");
+ scheduledTimeField.setAccessible(true);
+ scheduledTimeField.setLong(_connection, scheduledTime);
+ }
+}
diff --git
a/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionPlainDelegateTest.java
b/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionPlainDelegateTest.java
new file mode 100644
index 0000000000..3d445a28ce
--- /dev/null
+++
b/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionPlainDelegateTest.java
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.qpid.server.transport;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.Collection;
+import java.util.List;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.model.port.AmqpPort;
+import org.apache.qpid.test.utils.UnitTestBase;
+
+class NonBlockingConnectionPlainDelegateTest extends UnitTestBase
+{
+ private NonBlockingConnection _connection;
+ private NonBlockingConnectionPlainDelegate _delegate;
+
+ @BeforeEach
+ void beforeEach()
+ {
+ _connection = mock(NonBlockingConnection.class);
+ final AmqpPort<?> port = mock(AmqpPort.class);
+ when(port.getNetworkBufferSize()).thenReturn(1024);
+ _delegate = new NonBlockingConnectionPlainDelegate(_connection, port);
+ }
+
+ @AfterEach
+ void afterEach()
+ {
+ _delegate.shutdownInput();
+ }
+
+ @Test
+ void emptyCollectionIsComplete() throws Exception
+ {
+ final NonBlockingConnectionDelegate.WriteResult result =
_delegate.doWrite(List.of());
+
+ assertTrue(result.isComplete(), "Empty collection should be complete");
+ assertEquals(0, result.getBytesConsumed(), "Empty collection should
not consume bytes");
+ verify(_connection, never()).writeToTransport(any());
+ }
+
+ @Test
+ void partialWriteIsIncomplete() throws Exception
+ {
+ try (final QpidByteBuffer first = QpidByteBuffer.wrap(new byte[] { 1,
2 });
+ final QpidByteBuffer second = QpidByteBuffer.wrap(new byte[] { 3
}))
+ {
+ final List<QpidByteBuffer> buffers = List.of(first, second);
+ when(_connection.writeToTransport(buffers)).thenReturn(2L);
+
+ final NonBlockingConnectionDelegate.WriteResult result =
_delegate.doWrite(buffers);
+
+ assertFalse(result.isComplete(), "Buffers with remaining data
should be incomplete");
+ assertEquals(2, result.getBytesConsumed(), "Unexpected consumed
byte count");
+ }
+ }
+
+ @Test
+ void fullyConsumedWriteIsComplete() throws Exception
+ {
+ try (final QpidByteBuffer first = QpidByteBuffer.wrap(new byte[] { 1,
2 });
+ final QpidByteBuffer second = QpidByteBuffer.wrap(new byte[] { 3
}))
+ {
+ final List<QpidByteBuffer> buffers = List.of(first, second);
+ doAnswer(invocation ->
+ {
+ final Collection<QpidByteBuffer> writtenBuffers =
invocation.getArgument(0);
+ for (final QpidByteBuffer buffer : writtenBuffers)
+ {
+ buffer.position(buffer.limit());
+ }
+ return 3L;
+ }).when(_connection).writeToTransport(buffers);
+
+ final NonBlockingConnectionDelegate.WriteResult result =
_delegate.doWrite(buffers);
+
+ assertTrue(result.isComplete(), "Fully consumed buffers should be
complete");
+ assertEquals(3, result.getBytesConsumed(), "Unexpected consumed
byte count");
+ }
+ }
+}
diff --git
a/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionTest.java
b/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionTest.java
index b18416a8fd..4153fc4b3e 100644
---
a/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionTest.java
+++
b/broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionTest.java
@@ -96,6 +96,7 @@ class NonBlockingConnectionTest
when(broker.getEventLogger()).thenReturn(eventLogger);
when(port.getContextValue(Integer.class,
AmqpPort.FINAL_WRITE_THRESHOLD)).thenReturn(100);
when(port.getContextValue(Long.class,
AmqpPort.FINAL_WRITE_TIMEOUT)).thenReturn(100L);
+ when(port.getContextValue(Integer.class,
AmqpPort.MAX_GATHERING_WRITE_BUFFERS)).thenReturn(1024);
when(port.getParent()).thenReturn(broker);
final Set<TransportEncryption> encryptionSet =
Set.of(TransportEncryption.NONE);
diff --git
a/broker-core/src/test/java/org/apache/qpid/server/transport/TCPandSSLTransportTest.java
b/broker-core/src/test/java/org/apache/qpid/server/transport/TCPandSSLTransportTest.java
index 6024869117..014ca02752 100644
---
a/broker-core/src/test/java/org/apache/qpid/server/transport/TCPandSSLTransportTest.java
+++
b/broker-core/src/test/java/org/apache/qpid/server/transport/TCPandSSLTransportTest.java
@@ -212,6 +212,7 @@ public class TCPandSSLTransportTest extends UnitTestBase
when(port.getContextValue(Integer.class,
AmqpPort.PORT_DIAGNOSIS_OF_SSL_ENGINE_LOOPING_BREAK_THRESHOLD)).thenReturn(1005);
when(port.getContextValue(Integer.class,
AmqpPort.FINAL_WRITE_THRESHOLD)).thenReturn(100);
when(port.getContextValue(Long.class,
AmqpPort.FINAL_WRITE_TIMEOUT)).thenReturn(100L);
+ when(port.getContextValue(Integer.class,
AmqpPort.MAX_GATHERING_WRITE_BUFFERS)).thenReturn(1024);
final ObjectMapper mapper = new ObjectMapper();
final JavaType type =
mapper.getTypeFactory().constructCollectionType(List.class, String.class);
final List<String> allowList =
mapper.readValue(Broker.DEFAULT_SECURITY_TLS_PROTOCOL_ALLOW_LIST, type);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
