From 9b9a9df84e38395117dd4b8161dcee029420cb7d Mon Sep 17 00:00:00 2001 From: ImDanXie Date: Thu, 24 Sep 2026 22:17:49 +0800 Subject: [PATCH 1/2] fix(dist): assign the computed code to every recorded position for duplicate group routes When a batch contains two identical group MatchRoutes (protobuf value equality collapses them into one map entry, keeping only the later index), the earlier index remains null in the Code[] array and batchBuilder.addCode(code) throws NPE. The exception escapes before MutationResult is returned, but rangeWriter.done() still commits the buffered writes - leaving route cache / range fact inconsistent with the KV store until the cache expires. Normal routes are unaffected: that branch explicitly assigns codes[i] = OK for every index. The Group branch was missed. Record all positions for a route (List instead of Integer) and assign the computed code to every recorded index. The same change applies to batchRemoveRoute (delGroupMatchRecords). Regression test: testAddGroupRouteDuplicatedInOneBatchAllPositionsGetOk. Control experiment: fails on unfixed code (NPE), passes with the fix. Full DistWorkerCoProcTest (18) passes. --- .../bifromq/dist/worker/DistWorkerCoProc.java | 37 +++++++++------ .../dist/worker/DistWorkerCoProcTest.java | 46 +++++++++++++++++++ 2 files changed, 69 insertions(+), 14 deletions(-) diff --git a/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/DistWorkerCoProc.java b/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/DistWorkerCoProc.java index d49176e90..4d694936c 100644 --- a/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/DistWorkerCoProc.java +++ b/bifromq-dist/bifromq-dist-worker/src/main/java/org/apache/bifromq/dist/worker/DistWorkerCoProc.java @@ -311,7 +311,7 @@ private Runnable batchAddRoute(BatchMatchRequest request, replyBuilder.setReqId(request.getReqId()); Map normalRoutesAdded = new HashMap<>(); Map sharedRoutesAdded = new HashMap<>(); - Map> groupMatchRecords = new HashMap<>(); + Map>> groupMatchRecords = new HashMap<>(); Map resultMap = new HashMap<>(); request.getRequestsMap().forEach((tenantId, tenantMatchRequest) -> { BatchMatchReply.TenantBatch.Code[] codes = resultMap.computeIfAbsent(tenantId, @@ -351,7 +351,7 @@ private Runnable batchAddRoute(BatchMatchRequest request, ByteString groupRouteKey = toGroupRouteKey(tenantId, requestMatcher); RouteDetail routeDetail = RouteDetailCache.get(groupRouteKey); groupMatchRecords.computeIfAbsent(new GlobalTopicFilter(tenantId, routeDetail.matcher()), - k -> new HashMap<>()).put(route, i); + k -> new HashMap<>()).computeIfAbsent(route, key -> new ArrayList<>()).add(i); } } }); @@ -367,23 +367,28 @@ private Runnable batchAddRoute(BatchMatchRequest request, }); boolean updated = false; int maxMembers = request.getRequestsMap().get(tenantId).getOption().getMaxReceiversPerSharedSubGroup(); - for (MatchRoute route : newGroupMembers.keySet()) { - int resultIdx = newGroupMembers.get(route); + for (Map.Entry> entry : newGroupMembers.entrySet()) { + MatchRoute route = entry.getKey(); String receiverUrl = toReceiverUrl(route); + BatchMatchReply.TenantBatch.Code code; if (!matchGroup.containsMembers(receiverUrl)) { if (matchGroup.getMembersCount() < maxMembers) { matchGroup.putMembers(receiverUrl, route.getIncarnation()); - resultMap.get(tenantId)[resultIdx] = BatchMatchReply.TenantBatch.Code.OK; updated = true; + code = BatchMatchReply.TenantBatch.Code.OK; } else { - resultMap.get(tenantId)[resultIdx] = BatchMatchReply.TenantBatch.Code.EXCEED_LIMIT; + code = BatchMatchReply.TenantBatch.Code.EXCEED_LIMIT; } } else { if (matchGroup.getMembersMap().get(receiverUrl) < route.getIncarnation()) { matchGroup.putMembers(receiverUrl, route.getIncarnation()); updated = true; } - resultMap.get(tenantId)[resultIdx] = BatchMatchReply.TenantBatch.Code.OK; + code = BatchMatchReply.TenantBatch.Code.OK; + } + // a route repeated in the same batch shares one result across all its positions + for (int resultIdx : entry.getValue()) { + resultMap.get(tenantId)[resultIdx] = code; } } if (updated) { @@ -421,7 +426,7 @@ private Runnable batchRemoveRoute(BatchUnmatchRequest request, replyBuilder.setReqId(request.getReqId()); Map normalRoutesRemoved = new HashMap<>(); Map sharedRoutesRemoved = new HashMap<>(); - Map> delGroupMatchRecords = new HashMap<>(); + Map>> delGroupMatchRecords = new HashMap<>(); Map resultMap = new HashMap<>(); request.getRequestsMap().forEach((tenantId, tenantUnmatchRequest) -> { BatchUnmatchReply.TenantBatch.Code[] codes = resultMap.computeIfAbsent(tenantId, @@ -453,7 +458,7 @@ private Runnable batchRemoveRoute(BatchUnmatchRequest request, ByteString groupRouteKey = toGroupRouteKey(tenantId, requestMatcher); RouteDetail routeDetail = RouteDetailCache.get(groupRouteKey); delGroupMatchRecords.computeIfAbsent(new GlobalTopicFilter(tenantId, routeDetail.matcher()), - k -> new HashMap<>()).put(route, i); + k -> new HashMap<>()).computeIfAbsent(route, key -> new ArrayList<>()).add(i); } } }); @@ -467,13 +472,17 @@ private Runnable batchRemoveRoute(BatchUnmatchRequest request, assert matching instanceof GroupMatching; GroupMatching groupMatching = (GroupMatching) matching; Map existing = Maps.newHashMap(groupMatching.receivers()); - delGroupMembers.forEach((route, resultIdx) -> { + delGroupMembers.forEach((route, resultIdxs) -> { String receiverUrl = toReceiverUrl(route); + BatchUnmatchReply.TenantBatch.Code code; if (existing.containsKey(receiverUrl) && existing.get(receiverUrl) <= route.getIncarnation()) { existing.remove(receiverUrl); - resultMap.get(tenantId)[resultIdx] = BatchUnmatchReply.TenantBatch.Code.OK; + code = BatchUnmatchReply.TenantBatch.Code.OK; } else { - resultMap.get(tenantId)[resultIdx] = BatchUnmatchReply.TenantBatch.Code.NOT_EXISTED; + code = BatchUnmatchReply.TenantBatch.Code.NOT_EXISTED; + } + for (int resultIdx : resultIdxs) { + resultMap.get(tenantId)[resultIdx] = code; } }); if (existing.size() != groupMatching.receivers().size()) { @@ -492,8 +501,8 @@ private Runnable batchRemoveRoute(BatchUnmatchRequest request, .add(newGroupMatching); } } else { - delGroupMembers.forEach((detail, resultIdx) -> resultMap.get(tenantId)[resultIdx] = - BatchUnmatchReply.TenantBatch.Code.NOT_EXISTED); + delGroupMembers.forEach((detail, resultIdxs) -> resultIdxs.forEach(resultIdx -> + resultMap.get(tenantId)[resultIdx] = BatchUnmatchReply.TenantBatch.Code.NOT_EXISTED)); } }); resultMap.forEach((tenantId, codes) -> { diff --git a/bifromq-dist/bifromq-dist-worker/src/test/java/org/apache/bifromq/dist/worker/DistWorkerCoProcTest.java b/bifromq-dist/bifromq-dist-worker/src/test/java/org/apache/bifromq/dist/worker/DistWorkerCoProcTest.java index d862f1558..752aff0fe 100644 --- a/bifromq-dist/bifromq-dist-worker/src/test/java/org/apache/bifromq/dist/worker/DistWorkerCoProcTest.java +++ b/bifromq-dist/bifromq-dist-worker/src/test/java/org/apache/bifromq/dist/worker/DistWorkerCoProcTest.java @@ -587,6 +587,52 @@ public void testAddNormalRouteDuplicatedInOneBatchOnlyCountOnceAndRefresh() { assertEquals(reply.getResultsOrThrow(tenantId).getCode(1), BatchMatchReply.TenantBatch.Code.OK); } + @Test + public void testAddGroupRouteDuplicatedInOneBatchAllPositionsGetOk() { + long inc = 1L; + String tenantId = "tenantGD"; + String topicFilter = "$share/group/dup/topic"; + + BatchMatchRequest.TenantBatch tenantBatch = BatchMatchRequest.TenantBatch.newBuilder() + .setOption(TenantOption.newBuilder().setMaxReceiversPerSharedSubGroup(10).build()) + .addRoute(MatchRoute.newBuilder() + .setMatcher(TopicUtil.from(topicFilter)) + .setBrokerId(1) + .setReceiverId("inboxGD") + .setDelivererKey("delivererGD") + .setIncarnation(inc) + .build()) + .addRoute(MatchRoute.newBuilder() + .setMatcher(TopicUtil.from(topicFilter)) + .setBrokerId(1) + .setReceiverId("inboxGD") + .setDelivererKey("delivererGD") + .setIncarnation(inc) + .build()) + .build(); + + RWCoProcInput rwCoProcInput = RWCoProcInput.newBuilder().setDistService( + DistServiceRWCoProcInput.newBuilder() + .setBatchMatch(BatchMatchRequest.newBuilder() + .setReqId(3005) + .putRequests(tenantId, tenantBatch) + .build()) + .build()) + .build(); + + when(reader.get(any(ByteString.class))).thenReturn(Optional.empty()); + + Supplier resultSupplier = distWorkerCoProc.mutate(rwCoProcInput, reader, writer, + false); + IKVRangeCoProc.MutationResult result = resultSupplier.get(); + + // both positions must receive OK — the earlier index must not be silently null (NPE) + BatchMatchReply reply = result.output().getDistService().getBatchMatch(); + assertEquals(reply.getReqId(), 3005); + assertEquals(reply.getResultsOrThrow(tenantId).getCode(0), BatchMatchReply.TenantBatch.Code.OK); + assertEquals(reply.getResultsOrThrow(tenantId).getCode(1), BatchMatchReply.TenantBatch.Code.OK); + } + @Test public void testAddNormalRouteUpgradeIncNoIncButRefreshCalled() { long oldInc = 1L; From 2a9e2e63b7c4ac882d384e8bb53a0be69274b2c1 Mon Sep 17 00:00:00 2001 From: ImDanXie Date: Thu, 24 Sep 2026 14:40:38 +0000 Subject: [PATCH 2/2] test(dist): duplicate group routes in one batch must all receive OK (unmatch path) Symmetric regression test for batchRemoveRoute: two identical group MatchRoutes in the same tenant batch. The reader returns a RouteGroup protobuf containing the receiver so the removal path is exercised. Without the fix, the earlier index remains null and the NPE path is the same as in the add path. Control: both tests fail on unfixed code. --- .../dist/worker/DistWorkerCoProcTest.java | 52 +++++++++++++++++++ 1 file changed, 52 insertions(+) diff --git a/bifromq-dist/bifromq-dist-worker/src/test/java/org/apache/bifromq/dist/worker/DistWorkerCoProcTest.java b/bifromq-dist/bifromq-dist-worker/src/test/java/org/apache/bifromq/dist/worker/DistWorkerCoProcTest.java index 752aff0fe..df59f8b3f 100644 --- a/bifromq-dist/bifromq-dist-worker/src/test/java/org/apache/bifromq/dist/worker/DistWorkerCoProcTest.java +++ b/bifromq-dist/bifromq-dist-worker/src/test/java/org/apache/bifromq/dist/worker/DistWorkerCoProcTest.java @@ -633,6 +633,58 @@ public void testAddGroupRouteDuplicatedInOneBatchAllPositionsGetOk() { assertEquals(reply.getResultsOrThrow(tenantId).getCode(1), BatchMatchReply.TenantBatch.Code.OK); } + @Test + public void testRemoveGroupRouteDuplicatedInOneBatchAllPositionsGetOk() { + long inc = 1L; + String tenantId = "tenantGR"; + String topicFilter = "$share/group/rm/topic"; + + BatchUnmatchRequest.TenantBatch tenantBatch = BatchUnmatchRequest.TenantBatch.newBuilder() + .addRoute(MatchRoute.newBuilder() + .setMatcher(TopicUtil.from(topicFilter)) + .setBrokerId(1) + .setReceiverId("inboxGR") + .setDelivererKey("delivererGR") + .setIncarnation(inc) + .build()) + .addRoute(MatchRoute.newBuilder() + .setMatcher(TopicUtil.from(topicFilter)) + .setBrokerId(1) + .setReceiverId("inboxGR") + .setDelivererKey("delivererGR") + .setIncarnation(inc) + .build()) + .build(); + + RWCoProcInput rwCoProcInput = RWCoProcInput.newBuilder().setDistService( + DistServiceRWCoProcInput.newBuilder() + .setBatchUnmatch(BatchUnmatchRequest.newBuilder() + .setReqId(3006) + .putRequests(tenantId, tenantBatch) + .build()) + .build()) + .build(); + + // route exists in KV as a RouteGroup protobuf (receiverUrl -> incarnation) + // receiverUrl format: subBrokerId + NUL + receiverId + NUL + delivererKey + String receiverUrl = 1 + "\u0000" + "inboxGR" + "\u0000" + "delivererGR"; + org.apache.bifromq.dist.rpc.proto.RouteGroup group = + org.apache.bifromq.dist.rpc.proto.RouteGroup.newBuilder() + .putMembers(receiverUrl, inc) + .build(); + when(reader.get(any(ByteString.class))).thenReturn(Optional.of(group.toByteString())); + + Supplier resultSupplier = distWorkerCoProc.mutate(rwCoProcInput, reader, writer, + false); + IKVRangeCoProc.MutationResult result = resultSupplier.get(); + + // both positions must receive OK — the earlier index must not be silently null (NPE) + BatchUnmatchReply reply = result.output().getDistService().getBatchUnmatch(); + assertEquals(reply.getReqId(), 3006); + assertEquals(reply.getResultsOrThrow(tenantId).getCode(0), BatchUnmatchReply.TenantBatch.Code.OK); + assertEquals(reply.getResultsOrThrow(tenantId).getCode(1), BatchUnmatchReply.TenantBatch.Code.OK); + } + @Test public void testAddNormalRouteUpgradeIncNoIncButRefreshCalled() { long oldInc = 1L;