Summary

We started seeing the following in production:

java.net.SocketException: Broken pipe
	at java.net.SocketOutputStream.socketWrite0(Native Method)
	at java.net.SocketOutputStream.socketWrite(SocketOutputStream.java:109)
	at java.net.SocketOutputStream.write(SocketOutputStream.java:153)
	at java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:82)
	at java.io.BufferedOutputStream.flush(BufferedOutputStream.java:140)
	at java.io.DataOutputStream.flush(DataOutputStream.java:123)
	at com.rabbitmq.client.impl.SocketFrameHandler.flush(SocketFrameHandler.java:177)
	at com.rabbitmq.client.impl.AMQConnection.flush(AMQConnection.java:559)
	at com.rabbitmq.client.impl.AMQCommand.transmit(AMQCommand.java:127)
	at com.rabbitmq.client.impl.AMQChannel.quiescingTransmit(AMQChannel.java:396)
	at com.rabbitmq.client.impl.AMQChannel.transmit(AMQChannel.java:372)
	at com.rabbitmq.client.impl.AMQChannel.transmit(AMQChannel.java:365)
	at com.rabbitmq.client.impl.ChannelN.basicAck(ChannelN.java:1169)
	at com.rabbitmq.client.impl.recovery.RecoveryAwareChannelN.basicAck(RecoveryAwareChannelN.java:89)
	at com.rabbitmq.client.impl.recovery.AutorecoveringChannel.basicAck(AutorecoveringChannel.java:436)
	at RMQProblemTest$1.handleDelivery(RMQProblemTest.java:87)
	at com.rabbitmq.client.impl.ConsumerDispatcher$5.run(ConsumerDispatcher.java:149)
	at com.rabbitmq.client.impl.ConsumerWorkService$WorkPoolRunnable.run(ConsumerWorkService.java:100)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	at java.lang.Thread.run(Thread.java:745)

After looking at a packet capture, we were seeing [TCP Window Full] messages when RabbitMQ was sending to our consumer followed by a ton of [TCP ZeroWindow] frames coming from the consumer. After enough of these, RabbitMQ abruptly closes the connection by sending a RST frame and reporting only this in the logs:

=ERROR REPORT==== 10-Jan-2018::20:38:08 ===
closing AMQP connection <0.5373.1> ([::1]:53814 -> [::1]:5672):
{writer,send_failed,{error,timeout}}

Further investigation revealed that I was setting the QoS for the consumer only AFTER I had already started consuming (i.e. I was calling basicConsume directly before basicQos). I realize that this was wrong of me, however, it seems odd that this would cause the exception above especially with such a relatively small number of small messages. The more concerning thing is that none of the recovery/shutdown methods on either the connection or channel seemed to be called despite the fact that the connection was indeed closed.

Reproduction

I made a little test project that shows the issue.

https://github.com/vincentjames501/rmq-fail/blob/master/src/test/java/RMQProblemTest.java

Running the test will show the issue after 30 seconds or so. It appears reproducible on most RabbitMQ versions but I documented by setup below.

Environment

RabbitMQ server - 3.6.12 Erlang - Erlang 20.1 Operating system version (and distribution, if applicable) - OSX 10.13.2 All client libraries used - RabbitMQ Java Client 4.4.1 RabbitMQ plugins (if applicable) - Just Management Console

Thanks!

I described my problem here - https://github.com/rabbitmq/rabbitmq-java-client/issues/236 As per suggestion, I tried using v4.0.1 version of the library with NIO to work around the issue. The good thing is that that version does indeed detect connection problem much faster that 15 minutes that it used to take for previous version. The bad thing is that NIO main thread catches unhandled exception and terminates rendering the whole application unusable.

SEVERE: Unhandled exception. Aborting.
Throwable occurred: java.lang.AssertionError: BlockingCell can only be set once
	at com.rabbitmq.utility.BlockingCell.set (BlockingCell.java:141)
	at com.rabbitmq.client.impl.AMQConnection.doFinalShutdown (AMQConnection.java:681)
	at com.rabbitmq.client.impl.AMQConnection.handleHeartbeatFailure (AMQConnection.java:654)
	at com.rabbitmq.client.impl.nio.NioLoop.run (NioLoop.java:78)
	at java.lang.Thread.run (Unknown Source, bco=11)

I cannot easily reproduce it as I do not know what that bloody piece of network equipment does but it feels like it just stops outgoing packets from he client machine at some point while still allowing incoming ones.

previous exceptions that lead to the one above

Jan 16, 2017 11:45:56 AM com.rabbitmq.client.impl.nio.NioLoop run
WARNING: Error during reading frames
Throwable occurred: javax.net.ssl.SSLException: closed in read
	at com.rabbitmq.client.impl.nio.SslEngineByteBufferInputStream.read (SslEngineByteBufferInputStream.java:89)
	at java.io.DataInputStream.readUnsignedByte (Unknown Source, bco=4)
	at com.rabbitmq.client.impl.Frame.readFrom (Frame.java:91)
	at com.rabbitmq.client.impl.nio.NioLoop.run (NioLoop.java:149)
	at java.lang.Thread.run (Unknown Source, bco=11)
Jan 16, 2017 11:45:56 AM com.rabbitmq.client.impl.ForgivingExceptionHandler log
SEVERE: An unexpected connection driver error occured
Throwable occurred: com.rabbitmq.client.MissedHeartbeatException: Heartbeat missing with heartbeat = 30 seconds
	at com.rabbitmq.client.impl.AMQConnection.handleHeartbeatFailure (AMQConnection.java:648)
	at com.rabbitmq.client.impl.nio.NioLoop.run (NioLoop.java:78)
	at java.lang.Thread.run (Unknown Source, bco=11)
Jan 16, 2017 11:45:56 AM com.rabbitmq.client.impl.ForgivingExceptionHandler log
SEVERE: An unexpected connection driver error occured
Throwable occurred: javax.net.ssl.SSLException: closed in read
	at com.rabbitmq.client.impl.nio.SslEngineByteBufferInputStream.read (SslEngineByteBufferInputStream.java:89)
	at java.io.DataInputStream.readUnsignedByte (Unknown Source, bco=4)
	at com.rabbitmq.client.impl.Frame.readFrom (Frame.java:91)
	at com.rabbitmq.client.impl.nio.NioLoop.run (NioLoop.java:149)
	at java.lang.Thread.run (Unknown Source, bco=11)

sometime we need add more information to replyProperties (i.e set headers), so with handleCall function, we need replyPropertiesBuilder not replyProperties argument.

...in case an app wants to log something / set a flag / fire an event when a recovery has begun but before it has completed.

This particular implementation is just an idea / work in progress and hasn't been tested extensively, but I thought it easier to communicate the idea via code than a thousand words trying to describe it.

The major downfall to this approach is that it's a breaking API change to RecoveryListener, though I suspect application use of RecoveryListener is somewhat less-than-common. Another approach might be to make a subinterface of RecoveryListener that holds the extra method (though that's ugly and requires a bunch of instanceof checking when notifying), or changing RecoveryListener to be an abstract class with an empty handleRecoveryStarted method, though this approach isn't exactly pretty either.

Yet another approach would be to create a completely independent interface called RecoveryStartedListener and have them added separately, so they'd be managed independently. This would avoid the breaking API change, but it would add more bookkeeping to the code.

Our use case is that we'd both like to log when a recovery has started and set a flag that a recovery is in progress, then flip the bit back when recovery has completed (and continue to log it as we do now) so that our background task that tries very hard to ensure that consumers keep consuming over dodgy Internet connections can make more intelligent decisions about when to abort & replace a connection (sometimes necessary if an IP address changes, etc., but we'd rather give automatic recovery a chance to work first if it's in progress and likely to succeed).

Fixed #37 (at least to a large extent).

The tests can be run with mvn clean test or with ant, I kept it compatible.

And also with the right settings.xml in your ~/.m2 directory like

<settings xmlns="http://maven.apache.org/SETTINGS/1.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 xsi:schemaLocation="http://maven.apache.org/SETTINGS/1.0.0
                      http://maven.apache.org/xsd/settings-1.0.0.xsd">
<servers>
    <server>
      <id>sonatype-nexus-staging</id>
      <username>my_login</username>
      <password>my_password</password>
    </server>
  </servers>
</settings>

you can run mvn clean deploy that would compile, run the tests, and deploy to the repository if the tests pass.

I also added maven version plugin, so to set a version you can do it that way mvn versions:set -DnewVersion=<new.version> && mvn versions:commit

That's a better alternative than the release plugin. So basically to release a new version the maven command would be: mvn versions:set -DnewVersion=<new.version> && mvn versions:commit && mvn clean deploy

Version - amqp-client-3.6.1

When init is called on AutorecoveringConnection it adds the automatic recovery listener to the shutdownHooks list - Because this is the first entry, any user added shutdown hooks will be called after the automatic recovery.

as a possible solution; The listener created in addAutomaticRecoverListener would likely need to call each shutdownListener, and omit adding the created listener to the shutdownHooks. addShutdownListener, removeShutdownListener would only apply to the shutdownHooks list and not add to the delegate. recoverShutdownListeners would just call addAutomaticRecoverListener.

Hey there! This is a feature request I'd be willing to create PR for.

As it stands now, when you're publishing a message and it is successful, the published counter in MetricsCollector will be incremented. I'd like to add publishFailed counter, which will be incremented if and when publishing attempt fails.

This will allow users to easily create reports of publishing success rates and help them determine if the number messages they're not publishing are a significant enough to increase resiliency of their programs.

I'd like to find out if this is a feature that would be a welcome addition and if the way I'd like to add it would meet your expectations.

In all versions of this client, including the current version in master, the basic QoS "prefetch count" property is specified of type int. However, the AMQP 0.9.1 Reference specifies prefetch count to be of type short, which it further clarifies to be 16-bit integer.

This most naturally maps to java type short rather than int.

As a result, it's relatively simple to specify values that are out of bounds, such as 100000 (100k). These values have their most significant bits truncated and result in a transmitted value to the server of something else. In the case of 100000 the server will see the 16 least significant bits, which come out to value 16960. (See this SO post for context)

Possible fixes for this would include one or all of: a) explicitly failing of the value supplied is > Short.MAX_VALUE b) changing the allowed type in the client from int to short

When multiple channels share an AutorecoveringConnection and queues are coming and going from multiple theads, about 1 time out of 10000 , we are seeing a NullPointerException or ConcurrentModificationException to be thrown:

java.lang.NullPointerException
        at com.rabbitmq.client.impl.recovery.AutorecoveringConnection.removeBindingsWithDestination(AutorecoveringConnection.java:768)
        at com.rabbitmq.client.impl.recovery.AutorecoveringConnection.deleteRecordedQueue(AutorecoveringConnection.java:690)
        at com.rabbitmq.client.impl.recovery.AutorecoveringChannel.deleteRecordedQueue(AutorecoveringChannel.java:562)
        at com.rabbitmq.client.impl.recovery.AutorecoveringChannel.queueDelete(AutorecoveringChannel.java:303)
        at com.rabbitmq.client.impl.recovery.AutorecoveringChannel.queueDelete(AutorecoveringChannel.java:299)

The code at this location looks innocuous enough:

    Set<RecordedBinding> removeBindingsWithDestination(String s) {
        Set<RecordedBinding> result = new HashSet<RecordedBinding>();
        for (Iterator<RecordedBinding> it = this.recordedBindings.iterator(); it.hasNext(); ) {
            RecordedBinding b = it.next();
            if(b.getDestination().equals(s)) { // [[ NPE here ]] //
                it.remove();
                result.add(b);
            }
        }
        return result;
    }

However, removeBindingsWithDestination is not synchronized, so it's possible for multiple threads to be in this method changing the recordedBindings ArrayList at the same time as bindings come and go from multiple threads, even with channels not being shared across threads (but the Connection is shared).

It looks the other accesses to recordedBindings are synchronized (though caution: maybeDeleteRecordedAutoDeleteExchange is synchronizing differently than the others) so removeBindingsWithDestination may have just fallen through the cracks.

It may be a sufficient fix just to make removeBindingsWithDestination synchronized.

(Another tangential thing I noticed is that most of the recorded maintenance is done from synchronized methods, but maybeDeleteRecordedAutoDeleteQueue during a basicCancel instead synchronizes on recordedQueues and consumers, and that this method just removes the recordedQueues entry, not the recordedBindings associated with the queue as well.)

I found the following code in the class RecoveryAwareChannelN:

@Override
public void basicAck(long deliveryTag, boolean multiple) throws IOException {
    long realTag = deliveryTag - activeDeliveryTagOffset;
    // 0 tag means ack all when multiple is set
    if (realTag > 0 || (multiple && realTag == 0)) {
        transmit(new Basic.Ack(realTag, multiple));
        metricsCollector.basicAck(this, deliveryTag, multiple);
    }
}

When activeDeliveryTagOffset and deliveryTag are the same (this can happen directly after a channel is reconnected/recovered) this method will ack all outstanding messages instead of acking nothing.

The output of the attached sample application contains PRECONDITION_FAILED - unknown delivery tag 1 because it is trying to ack all messages. But it was already acked by acking the first message after connection recovery.

example-output.txt TestRabbitRecover.txt

• RabbitMQ version 3.6.15

• Erlang version Erlang 19.3.6.7

• Client library version (for all libraries used) com.rabbitmq/amqp-client/4.5.0

Because WorkPool uses an unbounded LinkedList, when a consumer is not fast enough the list will be filled by the connection thread until OOM error. This is aggravated by GC pressure making consumers even slower.

I suggest using a bounded blocking queue instead.

Hello,

We are using this version com.rabbitmq amqp-client 5.15.0

But 3.16.0 hasn't changed in the following regards.

We are facing the issue that we need a unique identifier that should change after each connection incl. automatic recovery. That's why we added this UUID into the clientProperties of the connection via ConnectionFactory. Everything works but the initial connection.created message the rabbit server is sending after a recovery. Here we see the old UUID still. We tried to change the property in the recoveryStarted call.

But when I look into the source code, we don't have any chance to do this.

AutorecoveringConnection stores params, which are not accessible from outside but in parts through a delegate that is created as RecoveryAwareAMQConnection using this params. Unfortunately they are stored as a new HashMap in AMQConnection: this._clientProperties = new HashMap(params.getClientProperties()); As well as whole as params in RecoveryAwareAMQConnectionFactory.

The AutorecoveringConnection.recoverConnection is calling the internal ConnectionFactory (not ours) to create a newConnection which is using the stored params, which is not accessible for us.

As I previously said, the first call of recoveryStart is too late for us to make any modification, because we need this changed clientProperty in the first communication (connection.created message) that is made before any channel activity.

Or is there a way to do this anyhow? In my small opinion the clientProperties of the delegate of the current connection should be used for the new connection instead of the initially stored params of the factory.

Kind regards, Markus Heidt

I know, I hate those cases as well :)

To avoid thread execution for too long, we interrupt the thread after the time limit is reached. Recently, We find that occasionally messages get lost. We use TX mode , and txCommit() got an ChannelContinuationTimeoutException. I found the following problem according to the warning log.

if thread interrupted before rpc(m,k) then rpc will not be sent to the server. And k.getReply(..) will hang for _rpcTimeout or forever.

client version is 5.8.0

com.rabbitmq.client.impl.AMQChannel

private AMQCommand privateRpc(Method m)
        throws IOException, ShutdownSignalException
    {
        SimpleBlockingRpcContinuation k = new SimpleBlockingRpcContinuation(m);
        rpc(m, k);
        // At this point, the request method has been sent, and we
        // should wait for the reply to arrive.
        //
        // Calling getReply() on the continuation puts us to sleep
        // until the connection's reader-thread throws the reply over
        // the fence or the RPC times out (if enabled)
        if(_rpcTimeout == NO_RPC_TIMEOUT) {
            return k.getReply();
        } else {
            try {
                return k.getReply(_rpcTimeout);
            } catch (TimeoutException e) {
                throw wrapTimeoutException(m, e);
            }
        }
    }

Several methods defined in ExceptionHandler interface accept Throwable as one of arguments. ForgivingExceptionHandler and DefaultExceptionHandler (which is used as a default ExceptionHandler implementation) as one of it's descendants effectively ignore any encountered Errors leaving application in abnormal state, which is strongly discouraged according to Error's description. Maybe it would be more correct to re-throw catched Errors after ExceptionHandler tried to log them or catch clauses should not even rely on ExceptionHandler to re-throw Errors.

We noticed that when using java client version 5.11.0 there is a significant skew in client connections towards the first Rabbit node in the list.

It appears that AutoRecovering connection actually shuffles the list of rabbit nodes when creating new connections, but regular connection does not.

public RecoveryAwareAMQConnection newConnection() throws IOException, TimeoutException {
    Exception lastException = null;
    List<Address> shuffled = shuffle(addressResolver.getAddresses());

There is no shuffle call in the regular ConnectionFactory.newConnection method.

Related to #330.

Many settings have been added to ConnectionFactory over the years, making it somewhat harder to configure. It's for instance easy to omit an important option when setting TLS, because the different TLS-related options are scattered: ConnectionFactory#useSslProtocol, ConnectionFactory#setSslContextFactory, ConnectionFactory#enableHostnameVerification. Same thing for NIO (mitigated by NioParams).

A more "use-case oriented", less JavaBeans-like way could improve the situation, e.g.:

cf.nio().executor(myNioExecutor).socketChannelConfigurator(configurator) // NIO configuration
  .connectionFactory() // going back to the connection factory (call may be avoided with syntax trick)
  .tls().context(sslContext).hostnameVerification(true); // TLS configuration

Not longer useful configuration methods would be deprecated and scheduled for removal in 7.0.0.