@Test public void getPartitionGroupClusterTest() throws Exception { TopicConfig topicConfig = nameServiceStub.getTopicConfig(TopicName.parse("test1")); GetPartitionGroupClusterRequest getPartitionGroupClusterRequest = new GetPartitionGroupClusterRequest(); Map<String, List<Integer>> groups = Maps.newHashMap(); groups.put(topicConfig.getName().getFullName(), Lists.newArrayList()); getPartitionGroupClusterRequest.setGroups(groups); for (Map.Entry<Integer, PartitionGroup> entry : topicConfig.getPartitionGroups().entrySet()) { groups.get(topicConfig.getName().getFullName()).add(entry.getKey()); } groups.get(topicConfig.getName().getFullName()).add(1); ByteBuf byteBuf = ByteBufAllocator.DEFAULT.heapBuffer(); new GetPartitionGroupClusterRequestCodec().encode(getPartitionGroupClusterRequest, byteBuf); getPartitionGroupClusterRequest = (GetPartitionGroupClusterRequest) new GetPartitionGroupClusterRequestCodec().decode(null, byteBuf); GetPartitionGroupClusterResponse response = (GetPartitionGroupClusterResponse) getPartitionGroupClusterRequestHandler.handle(null, new JoyQueueCommand(getPartitionGroupClusterRequest)).getPayload(); byteBuf = ByteBufAllocator.DEFAULT.heapBuffer(); new GetPartitionGroupClusterResponseCodec().encode(response, byteBuf); response = (GetPartitionGroupClusterResponse) new GetPartitionGroupClusterResponseCodec().decode(null, byteBuf); Assert.assertEquals(topicConfig.getPartitionGroups().size(), response.getGroups().get(topicConfig.getName().getFullName()).size()); for (Map.Entry<Integer, PartitionGroup> entry : topicConfig.getPartitionGroups().entrySet()) { GetPartitionGroupClusterResponse.PartitionGroupCluster cluster = response.getCluster(topicConfig.getName().getFullName(), entry.getKey()); Assert.assertEquals((int) entry.getValue().getLeader(), cluster.getRWNode().getId()); } }
@Override public Command handle(Transport transport, Command command) { GetPartitionGroupClusterRequest request = (GetPartitionGroupClusterRequest) command.getPayload(); GetPartitionGroupClusterResponse response = new GetPartitionGroupClusterResponse(); for (Map.Entry<String, List<Integer>> entry : request.getGroups().entrySet()) { String topic = entry.getKey(); for (Integer partitionGroup : entry.getValue()) { StoreNodes nodes = storeService.getNodes(topic, partitionGroup); if (nodes == null) { continue; } GetPartitionGroupClusterResponse.PartitionGroupCluster partitionGroupCluster = new GetPartitionGroupClusterResponse.PartitionGroupCluster(); for (StoreNode node : nodes.getNodes()) { partitionGroupCluster.addNode(new GetPartitionGroupClusterResponse.PartitionGroupNode(node.getId(), node.isWritable(), node.isReadable())); } response.addCluster(topic, partitionGroup, partitionGroupCluster); } } return new Command(response); }
GetPartitionGroupClusterRequestHandler implements Type, BrokerCommandHandler, BrokerContextAware { @Override public Command handle(Transport transport, Command command) { GetPartitionGroupClusterRequest request = (GetPartitionGroupClusterRequest) command.getPayload(); GetPartitionGroupClusterResponse response = new GetPartitionGroupClusterResponse(); for (Map.Entry<String, List<Integer>> entry : request.getGroups().entrySet()) { String topic = entry.getKey(); for (Integer partitionGroup : entry.getValue()) { StoreNodes nodes = storeService.getNodes(topic, partitionGroup); if (nodes == null) { continue; } GetPartitionGroupClusterResponse.PartitionGroupCluster partitionGroupCluster = new GetPartitionGroupClusterResponse.PartitionGroupCluster(); for (StoreNode node : nodes.getNodes()) { partitionGroupCluster.addNode(new GetPartitionGroupClusterResponse.PartitionGroupNode(node.getId(), node.isWritable(), node.isReadable())); } response.addCluster(topic, partitionGroup, partitionGroupCluster); } } return new Command(response); } }
GetPartitionGroupClusterRequestHandler implements Type, BrokerCommandHandler, BrokerContextAware { @Override public Command handle(Transport transport, Command command) { GetPartitionGroupClusterRequest request = (GetPartitionGroupClusterRequest) command.getPayload(); GetPartitionGroupClusterResponse response = new GetPartitionGroupClusterResponse(); for (Map.Entry<String, List<Integer>> entry : request.getGroups().entrySet()) { String topic = entry.getKey(); for (Integer partitionGroup : entry.getValue()) { StoreNodes nodes = storeService.getNodes(topic, partitionGroup); if (nodes == null) { continue; } GetPartitionGroupClusterResponse.PartitionGroupCluster partitionGroupCluster = new GetPartitionGroupClusterResponse.PartitionGroupCluster(); for (StoreNode node : nodes.getNodes()) { partitionGroupCluster.addNode(new GetPartitionGroupClusterResponse.PartitionGroupNode(node.getId(), node.isWritable(), node.isReadable())); } response.addCluster(topic, partitionGroup, partitionGroupCluster); } } return new Command(response); }  }
GetPartitionGroupClusterRequestHandler implements Type, BrokerCommandHandler, BrokerContextAware { @Override public Command handle(Transport transport, Command command) { GetPartitionGroupClusterRequest request = (GetPartitionGroupClusterRequest) command.getPayload(); GetPartitionGroupClusterResponse response = new GetPartitionGroupClusterResponse(); for (Map.Entry<String, List<Integer>> entry : request.getGroups().entrySet()) { String topic = entry.getKey(); for (Integer partitionGroup : entry.getValue()) { StoreNodes nodes = storeService.getNodes(topic, partitionGroup); if (nodes == null) { continue; } GetPartitionGroupClusterResponse.PartitionGroupCluster partitionGroupCluster = new GetPartitionGroupClusterResponse.PartitionGroupCluster(); for (StoreNode node : nodes.getNodes()) { partitionGroupCluster.addNode(new GetPartitionGroupClusterResponse.PartitionGroupNode(node.getId(), node.isWritable(), node.isReadable())); } response.addCluster(topic, partitionGroup, partitionGroupCluster); } } return new Command(response); }  @Override void setBrokerContext(BrokerContext brokerContext); @Override Command handle(Transport transport, Command command); @Override int type(); }
GetPartitionGroupClusterRequestHandler implements Type, BrokerCommandHandler, BrokerContextAware { @Override public Command handle(Transport transport, Command command) { GetPartitionGroupClusterRequest request = (GetPartitionGroupClusterRequest) command.getPayload(); GetPartitionGroupClusterResponse response = new GetPartitionGroupClusterResponse(); for (Map.Entry<String, List<Integer>> entry : request.getGroups().entrySet()) { String topic = entry.getKey(); for (Integer partitionGroup : entry.getValue()) { StoreNodes nodes = storeService.getNodes(topic, partitionGroup); if (nodes == null) { continue; } GetPartitionGroupClusterResponse.PartitionGroupCluster partitionGroupCluster = new GetPartitionGroupClusterResponse.PartitionGroupCluster(); for (StoreNode node : nodes.getNodes()) { partitionGroupCluster.addNode(new GetPartitionGroupClusterResponse.PartitionGroupNode(node.getId(), node.isWritable(), node.isReadable())); } response.addCluster(topic, partitionGroup, partitionGroupCluster); } } return new Command(response); }  @Override void setBrokerContext(BrokerContext brokerContext); @Override Command handle(Transport transport, Command command); @Override int type();  }