Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

ARTEMIS-3392 Scale down would fail if target queue's id greater than max int #3661

Merged
merged 1 commit into from Jul 21, 2021
Merged

ARTEMIS-3392 Scale down would fail if target queue's id greater than max int #3661

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
6 changes: 3 additions & 3 deletions ...s-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ScaleDownHandler.java
View file
Open in desktop
Expand Up @@ -451,8 +451,8 @@ private long createQueueWithRoutingTypeIfNecessaryAndGetID(ClientSession session
return queueID;
}

private Integer getQueueID(ClientSession session, SimpleString queueName) throws Exception {
Integer queueID = -1;
private Long getQueueID(ClientSession session, SimpleString queueName) throws Exception {
Long queueID = -1L;
Object result;
try (ClientRequestor requestor = new ClientRequestor(session, "activemq.management")) {
ClientMessage managementMessage = session.createMessage(false);
Expand All @@ -463,7 +463,7 @@ private Integer getQueueID(ClientSession session, SimpleString queueName) throws
result = ManagementHelper.getResult(reply);
}
if (result != null && result instanceof Number) {
queueID = ((Number) result).intValue();
queueID = ((Number) result).longValue();
}
return queueID;
}
Expand Down
51 changes: 51 additions & 0 deletions ...c/test/java/org/apache/activemq/artemis/tests/integration/server/ScaleDownDirectTest.java
View file
Open in desktop
Expand Up @@ -29,6 +29,8 @@
import org.apache.activemq.artemis.api.core.client.ClientProducer;
import org.apache.activemq.artemis.api.core.client.ClientSession;
import org.apache.activemq.artemis.api.core.client.ClientSessionFactory;
import org.apache.activemq.artemis.core.persistence.impl.journal.BatchingIDGenerator;
import org.apache.activemq.artemis.core.persistence.impl.journal.JournalStorageManager;
import org.apache.activemq.artemis.core.postoffice.impl.LocalQueueBinding;
import org.apache.activemq.artemis.core.server.impl.ScaleDownHandler;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
Expand Down Expand Up @@ -294,6 +296,55 @@ public void testTemporaryQueues() throws Exception {
Assert.assertNull(servers[1].getPostOffice().getBinding(new SimpleString(queueName3)));
}

@Test
public void testScaleDownWithBigQueueID() throws Exception {
final int TEST_SIZE = 2;
final String addressName = "testAddress";
final String queueName1 = "testQueue1";

JournalStorageManager manager = (JournalStorageManager) servers[0].getStorageManager();
BatchingIDGenerator idGenerator = (BatchingIDGenerator) manager.getIDGenerator();
idGenerator.forceNextID((Integer.MAX_VALUE) + 100L);

long nextId = idGenerator.generateID();
assertTrue(nextId > Integer.MAX_VALUE);

manager = (JournalStorageManager) servers[1].getStorageManager();
idGenerator = (BatchingIDGenerator) manager.getIDGenerator();
idGenerator.forceNextID((Integer.MAX_VALUE) + 200L);

nextId = idGenerator.generateID();
assertTrue(nextId > Integer.MAX_VALUE);

// create 2 queues on each node mapped to the same address
createQueue(0, addressName, queueName1, null, true);
createQueue(1, addressName, queueName1, null, true);

// send messages to node 0
send(0, addressName, TEST_SIZE, true, null);

// at this point on node 0 there should be 2 messages in testQueue1
Wait.assertEquals(TEST_SIZE, () -> getMessageCount(((LocalQueueBinding) servers[0].getPostOffice().getBinding(new SimpleString(queueName1))).getQueue()));

assertEquals(TEST_SIZE, performScaledown());
// trigger scaleDown from node 0 to node 1
servers[0].stop();

// get the 2 messages from queue 1
addConsumer(0, 1, queueName1, null);
ClientMessage clientMessage = consumers[0].getConsumer().receive(250);
Assert.assertNotNull(clientMessage);
clientMessage.acknowledge();
clientMessage = consumers[0].getConsumer().receive(250);
Assert.assertNotNull(clientMessage);
clientMessage.acknowledge();

// ensure there are no more messages on queue 1
clientMessage = consumers[0].getConsumer().receive(250);
Assert.assertNull(clientMessage);
removeConsumer(0);
}

private void checkBody(ClientMessage message, int bufferSize) {
assertEquals(bufferSize, message.getBodySize());
byte[] body = new byte[message.getBodySize()];
Expand Down