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..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 @@ -587,6 +587,104 @@ 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 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;