Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

AMQP-463: Fix Publisher Confirms with JRockit

Browse Source 
JIRA: https://jira.spring.io/browse/AMQP-463

JRockit throws an `IllegalArgumentException` if a
`TreeMap$Entry` is used after removal via the `Iterator`.

The `PublishSubscribeChannelImpl` uses this technique.

Obtain the entry value before removing the entry.

Add a test to reproduce the problem with the JRockit JVM.

Currently sends 10k messages across 100 threads; also tested
with 1m messages across 100 threads with no lost confirms.
This commit is contained in:
Gary Russell
2015-01-05 13:30:34 -05:00
parent 1f38cee752
commit ca14e276c6
3 changed files with 30 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -723,8 +723,9 @@ public class PublisherCallbackChannelImpl
Iterator<Entry<Long, PendingConfirm>> iterator = confirms.entrySet().iterator();
while (iterator.hasNext()) {
Entry<Long, PendingConfirm> entry = iterator.next();
PendingConfirm value = entry.getValue();
iterator.remove();
doHandleConfirm(ack, involvedListener, entry.getValue());
doHandleConfirm(ack, involvedListener, value);
}
}
}

4
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitAdminDeclarationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -278,7 +278,7 @@ public class RabbitAdminDeclarationTests {
verify(Config.channel2, times(0)).queueDeclare("foo", true, false, false, null);
verify(Config.channel2, times(0)).exchangeDeclare("bar", "direct", true, false, new HashMap<String, Object>());
verify(Config.channel2, times(0)).queueBind("foo", "bar", "foo", null);
context.destroy();
context.close();
}
@Test

32
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2010-2014 the original author or authors.
* Copyright 2010-2015 the original author or authors.
*
* Licensed 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
@@ -100,13 +100,13 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
public void create() {
connectionFactory = new CachingConnectionFactory();
connectionFactory.setHost("localhost");
connectionFactory.setChannelCacheSize(1);
connectionFactory.setChannelCacheSize(10);
connectionFactory.setPort(BrokerTestUtils.getPort());
connectionFactoryWithConfirmsEnabled = new CachingConnectionFactory();
connectionFactoryWithConfirmsEnabled.setHost("localhost");
// When using publisher confirms, the cache size needs to be large enough
// otherwise channels can be closed before confirms are received.
connectionFactoryWithConfirmsEnabled.setChannelCacheSize(10);
connectionFactoryWithConfirmsEnabled.setChannelCacheSize(100);
connectionFactoryWithConfirmsEnabled.setPort(BrokerTestUtils.getPort());
connectionFactoryWithConfirmsEnabled.setPublisherConfirms(true);
templateWithConfirmsEnabled = new RabbitTemplate(connectionFactoryWithConfirmsEnabled);
@@ -138,18 +138,36 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
@Test
public void testPublisherConfirmReceived() throws Exception {
final CountDownLatch latch = new CountDownLatch(10);
final CountDownLatch latch = new CountDownLatch(10000);
final AtomicInteger acks = new AtomicInteger();
templateWithConfirmsEnabled.setConfirmCallback(new ConfirmCallback() {
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
acks.incrementAndGet();
latch.countDown();
}
});
for (int i = 0; i < 10; i++) {
templateWithConfirmsEnabled.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc"));
ExecutorService exec = Executors.newCachedThreadPool();
for (int i = 0; i < 100; i++) {
exec.submit(new Runnable() {
@Override
public void run() {
try {
for (int i = 0; i < 100; i++) {
templateWithConfirmsEnabled.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc"));
}
}
catch (Throwable t) {
t.printStackTrace();
}
}
});
}
assertTrue(latch.await(10, TimeUnit.SECONDS));
exec.shutdown();
assertTrue(exec.awaitTermination(300, TimeUnit.SECONDS));
assertTrue("" + latch.getCount(), latch.await(60, TimeUnit.SECONDS));
assertNull(templateWithConfirmsEnabled.getUnconfirmed(-1));
this.templateWithConfirmsEnabled.execute(new ChannelCallback<Void>() {