Skip to content

UT: add test for queue lookup by appId/subQId #141

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

Open
678098 wants to merge 2 commits into
base: main
Choose a base branch
from
Open

UT: add test for queue lookup by appId/subQId #141

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ce1baa4
UT: add test for queue lookup by appId/subQId
678098 Sep 18, 2026
a866ede
Fix: scan expired queues by both qId/appId
678098 Sep 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion bmq-sdk/src/main/java/com/bloomberg/bmq/impl/LateResponseHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,7 +106,7 @@ private void handleConfigureStreamResponse(ConfigureStreamResponse configureStre
}

int qId = originalRequest.id();
Integer subQId = queueStateManager.findSubQId(parameters.appId());
Integer subQId = queueStateManager.findSubQId(qId, parameters.appId());
if (subQId == null) {
subQId = QueueId.k_DEFAULT_SUBQUEUE_ID;
}
Expand Down
52 changes: 31 additions & 21 deletions bmq-sdk/src/main/java/com/bloomberg/bmq/impl/QueueManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.TreeMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
Expand Down Expand Up @@ -49,7 +50,6 @@ public class QueueManager {
private Map<QueueId, QueueImpl> expiredQueueMap;

private Map<Integer, QueueImpl> subscriptionIdMap;
private Map<String, Integer> appId_subQId_Map;
private AtomicInteger nextQueueId;

private QueueManager() {
Expand All @@ -58,7 +58,6 @@ private QueueManager() {
keyQueueIdMap = new HashMap<>();
expiredQueueMap = new HashMap<>();
subscriptionIdMap = new HashMap<>();
appId_subQId_Map = new HashMap<>();
lock = new Object();
nextQueueId = new AtomicInteger(0);
}
Expand All @@ -71,7 +70,6 @@ public void reset() {
keyQueueIdMap = new HashMap<>();
expiredQueueMap = new HashMap<>();
subscriptionIdMap = new HashMap<>();
appId_subQId_Map = new HashMap<>();
nextQueueId = new AtomicInteger(0);
}
}
Expand Down Expand Up @@ -147,14 +145,6 @@ public boolean insertExpired(QueueImpl queue) {
return false;
}
expiredQueueMap.put(queueId, queue);

SubQueueIdInfo subQueueIdInfo = queue.getParameters().getSubIdInfo();
if (subQueueIdInfo != null) {
String appId = subQueueIdInfo.appId();
if (appId != null) {
appId_subQId_Map.put(appId, queueId.getSubQId());
}
}
}
return true;
}
Expand Down Expand Up @@ -236,14 +226,6 @@ public boolean removeExpired(QueueImpl queue) {
"Wrong QueueIds: %d != %d",
queue.getQueueId(), qHandle.getQueueId()));
}

SubQueueIdInfo subQueueIdInfo = queue.getParameters().getSubIdInfo();
if (subQueueIdInfo != null) {
String appId = subQueueIdInfo.appId();
if (appId != null) {
appId_subQId_Map.remove(appId);
}
}
}
return true;
}
Expand Down Expand Up @@ -294,9 +276,37 @@ public QueueImpl findByUri(Uri uri) {
}
}

public Integer findSubQId(String appId) {
/**
* Find subQueueId of an expired queue by queue id and appId.
*
* <p>All subQueues of one canonical URI share the queue id, and each of them has its own appId,
* so the two together identify the queue. The same appId may be reopened with a greater
* subQueueId while the previous one is still expired; in that case the oldest subQueueId is
* returned, as responses come in the order of the requests.
*
* <p>Thread safe.
*
* @param qId queue id used for search
* @param appId appId used for search
* @return subQueueId if the queue is found, otherwise null.
*/
public Integer findSubQId(int qId, String appId) {
synchronized (lock) {
return appId_subQId_Map.get(appId);
Integer subQId = null;
for (Map.Entry<QueueId, QueueImpl> entry : expiredQueueMap.entrySet()) {
QueueId queueId = entry.getKey();
if (queueId.getQId() != qId) {
continue;
}
SubQueueIdInfo subQueueIdInfo = entry.getValue().getParameters().getSubIdInfo();
if (subQueueIdInfo == null || !Objects.equals(subQueueIdInfo.appId(), appId)) {
continue;
}
if (subQId == null || queueId.getSubQId() < subQId) {
subQId = queueId.getSubQId();
}
}
return subQId;
}
}

Expand Down
4 changes: 2 additions & 2 deletions bmq-sdk/src/main/java/com/bloomberg/bmq/impl/QueueStateManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -225,8 +225,8 @@ private QueueStateManager() {
queueManager = QueueManager.createInstance();
}

public Integer findSubQId(String appId) {
return queueManager.findSubQId(appId);
public Integer findSubQId(int qId, String appId) {
return queueManager.findSubQId(qId, appId);
}

public QueueImpl findByQueueId(QueueId queueId) {
Expand Down
74 changes: 74 additions & 0 deletions bmq-sdk/src/test/java/com/bloomberg/bmq/impl/QueueManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
Expand Down Expand Up @@ -279,6 +280,79 @@ void removeExpiredQueueTest() {
logger.info("==================================================================");
}

/**
* Test that an expired queue is looked up by both its canonical URI and its appId.
*
* <p>The same appId may be used by queues with different canonical URIs, and each of them has
* its own subQueueId. {@link LateResponseHandler} rebuilds a {@link QueueId} from the qId of a
* late configure response and the subQueueId found by appId, so the lookup must give the
* subQueueId of the queue the response belongs to.
*
* <p>Test steps:
*
* <ol>
* <li>create queue manager instance
* <li>insert two active queues of one canonical URI with appIds "foo" and "bar", so that
* "bar" gets subQueueId 2
* <li>insert an active queue of another canonical URI with appId "bar", which gets subQueueId
* 1
* <li>move all three queues to expired queues, as a local timeout does
* <li>for each queue, rebuild a QueueId from its qId and the subQueueId found by its appId,
* the way LateResponseHandler does
* <li>check that every rebuilt QueueId resolves to the queue it was built from
* </ol>
*/
@Test
void lookupByUriAndAppIdTest() {
logger.info("====================================================");
logger.info("BEGIN Testing QueueManager lookupByUriAndAppIdTest.");
logger.info("====================================================");

QueueManager obj = QueueManager.createInstance();

Uri fooOfQueueA = new Uri("bmq://ts.trades.myapp/queueA?id=foo");
Uri barOfQueueA = new Uri("bmq://ts.trades.myapp/queueA?id=bar");
Uri barOfQueueB = new Uri("bmq://ts.trades.myapp/queueB?id=bar");

QueueImpl queueAFoo = insertActive(obj, fooOfQueueA);
QueueImpl queueABar = insertActive(obj, barOfQueueA);
QueueImpl queueBBar = insertActive(obj, barOfQueueB);

// Two appIds of one canonical URI, so the second one gets subQueueId 2.
assertEquals(1, queueAFoo.getSubQueueId());
assertEquals(2, queueABar.getSubQueueId());

// Another canonical URI starts its subQueueIds anew.
assertEquals(1, queueBBar.getSubQueueId());

assertTrue(obj.insertExpired(queueAFoo));
assertTrue(obj.insertExpired(queueABar));
assertTrue(obj.insertExpired(queueBBar));

assertEquals(queueAFoo, findExpiredByAppId(obj, queueAFoo));
assertEquals(queueABar, findExpiredByAppId(obj, queueABar));
assertEquals(queueBBar, findExpiredByAppId(obj, queueBBar));

logger.info("==================================================");
logger.info("END Testing QueueManager lookupByUriAndAppIdTest.");
logger.info("==================================================");
}

@678098 678098 Sep 18, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fails in main:

[ERROR] com.bloomberg.bmq.impl.QueueManagerTest.lookupByUriAndAppIdTest -- Time elapsed: 0.002 s <<< FAILURE!
org.opentest4j.AssertionFailedError: expected: <com.bloomberg.bmq.impl.QueueImpl@7306f65f> but was: <com.bloomberg.bmq.impl.QueueImpl@20309365>
	at org.junit.jupiter.api.Assertions.assertEquals(Assertions.java:1199)
	at com.bloomberg.bmq.impl.QueueManagerTest.lookupByUriAndAppIdTest(QueueManagerTest.java:333)

[INFO] Running com.bloomberg.bmq.impl.QueueManagerTest
[ERROR] Tests run: 1, Failures: 1, Errors: 0, Skipped: 0, Time elapsed: 0.002 s <<< FAILURE! -- in com.bloomberg.bmq.impl.QueueManagerTest
[ERROR] com.bloomberg.bmq.impl.QueueManagerTest.lookupByUriAndAppIdTest -- Time elapsed: 0.001 s <<< FAILURE!
org.opentest4j.AssertionFailedError: expected: <com.bloomberg.bmq.impl.QueueImpl@23043ba> but was: <com.bloomberg.bmq.impl.QueueImpl@51ce8293>
	at org.junit.jupiter.api.Assertions.assertEquals(Assertions.java:1199)
	at com.bloomberg.bmq.impl.QueueManagerTest.lookupByUriAndAppIdTest(QueueManagerTest.java:333)

[INFO] Running com.bloomberg.bmq.impl.QueueManagerTest
[ERROR] Tests run: 1, Failures: 1, Errors: 0, Skipped: 0, Time elapsed: 0.002 s <<< FAILURE! -- in com.bloomberg.bmq.impl.QueueManagerTest
[ERROR] com.bloomberg.bmq.impl.QueueManagerTest.lookupByUriAndAppIdTest -- Time elapsed: 0.001 s <<< FAILURE!
org.opentest4j.AssertionFailedError: expected: <com.bloomberg.bmq.impl.QueueImpl@400cb9ef> but was: <com.bloomberg.bmq.impl.QueueImpl@79fa9f97>
	at org.junit.jupiter.api.Assertions.assertEquals(Assertions.java:1199)
	at com.bloomberg.bmq.impl.QueueManagerTest.lookupByUriAndAppIdTest(QueueManagerTest.java:333)

[INFO] 
[INFO] Results:
[INFO] 
[ERROR] Failures: 
[ERROR] com.bloomberg.bmq.impl.QueueManagerTest.lookupByUriAndAppIdTest
[ERROR]   Run 1: QueueManagerTest.lookupByUriAndAppIdTest:333 expected: <com.bloomberg.bmq.impl.QueueImpl@52fe87e0> but was: <com.bloomberg.bmq.impl.QueueImpl@2fce8243>
[ERROR]   Run 2: QueueManagerTest.lookupByUriAndAppIdTest:333 expected: <com.bloomberg.bmq.impl.QueueImpl@7306f65f> but was: <com.bloomberg.bmq.impl.QueueImpl@20309365>
[ERROR]   Run 3: QueueManagerTest.lookupByUriAndAppIdTest:333 expected: <com.bloomberg.bmq.impl.QueueImpl@23043ba> but was: <com.bloomberg.bmq.impl.QueueImpl@51ce8293>
[ERROR]   Run 4: QueueManagerTest.lookupByUriAndAppIdTest:333 expected: <com.bloomberg.bmq.impl.QueueImpl@400cb9ef> but was: <com.bloomberg.bmq.impl.QueueImpl@79fa9f97>


private QueueImpl insertActive(QueueManager manager, Uri uri) {
QueueImpl queue = createQueue(session, uri, 0L);
QueueId queueId = manager.generateNextQueueId(uri);
queue.setQueueId(queueId.getQId()).setSubQueueId(queueId.getSubQId());
assertTrue(manager.insert(queue));
return queue;
}

/** Resolve an expired queue the way LateResponseHandler does. */
private QueueImpl findExpiredByAppId(QueueManager manager, QueueImpl queue) {
Integer subQId = manager.findSubQId(queue.getQueueId(), queue.getUri().id());
assertNotNull(subQId);
return manager.findExpiredByQueueId(QueueId.createInstance(queue.getQueueId(), subQId));
}

/**
* Test for queue manager to test increment queue substream count
*
Expand Down
Loading