@Test public void helixVcrClusterTest() throws Exception { StrictMatchExternalViewVerifier helixBalanceVerifier = new StrictMatchExternalViewVerifier(ZK_CONNECT_STRING, VCR_CLUSTER_NAME, Collections.singleton(VcrTestUtil.helixResource), null); VirtualReplicatorCluster helixInstance1 = createHelixInstance(8123, 10123); List<PartitionId> expectedPartitions = mockClusterMap.getAllPartitionIds(null); MockVcrListener mockVcrListener = new MockVcrListener(); helixInstance1.addListener(mockVcrListener); helixInstance1.participate(); TestUtils.checkAndSleep(true, () -> helixInstance1.getAssignedPartitionIds().size() > 0, 1000); Assert.assertTrue("Helix balance timeout.", helixBalanceVerifier.verify(5000)); Assert.assertEquals("Partition assignment are not correct.", helixInstance1.getAssignedPartitionIds(), expectedPartitions); VirtualReplicatorCluster helixInstance2 = createHelixInstance(8124, 10124); helixInstance2.participate(); TestUtils.checkAndSleep(true, () -> helixInstance1.getAssignedPartitionIds().size() < expectedPartitions.size(), 1000); Assert.assertTrue("Helix balance timeout.", helixBalanceVerifier.verify(5000)); Assert.assertEquals("Number of partitions removed are not correct.", expectedPartitions.size() / 2, mockVcrListener.getPartitionSet().size()); helixInstance2.close(); TestUtils.checkAndSleep(true, () -> helixInstance1.getAssignedPartitionIds().size() > expectedPartitions.size() / 2, 500); Assert.assertTrue("Helix balance timeout.", helixBalanceVerifier.verify(5000)); Assert.assertEquals("Partition assignment are not correct.", helixInstance1.getAssignedPartitionIds(), expectedPartitions); helixInstance1.close(); }
public HelixVcrCluster(CloudConfig cloudConfig, ClusterMapConfig clusterMapConfig, ClusterMap clusterMap) { if (Utils.isNullOrEmpty(cloudConfig.vcrClusterZkConnectString)) { throw new IllegalArgumentException("Missing value for " + CloudConfig.VCR_CLUSTER_ZK_CONNECT_STRING); } else if (Utils.isNullOrEmpty(cloudConfig.vcrClusterName)) { throw new IllegalArgumentException("Missing value for " + CloudConfig.VCR_CLUSTER_NAME); } this.cloudConfig = cloudConfig; currentDataNode = new CloudDataNode(cloudConfig, clusterMapConfig); List<? extends PartitionId> allPartitions = clusterMap.getAllPartitionIds(null); logger.info("All partitions from clusterMap: {}.", allPartitions); partitionIdMap = allPartitions.stream().collect(Collectors.toMap(PartitionId::toPathString, Function.identity())); vcrClusterName = cloudConfig.vcrClusterName; vcrInstanceName = ClusterMapUtils.getInstanceName(clusterMapConfig.clusterMapHostName, clusterMapConfig.clusterMapPort); metrics = new HelixVcrClusterMetrics(clusterMap.getMetricRegistry(), assignedPartitionIds); }
HelixVcrCluster implements VirtualReplicatorCluster { public HelixVcrCluster(CloudConfig cloudConfig, ClusterMapConfig clusterMapConfig, ClusterMap clusterMap) { if (Utils.isNullOrEmpty(cloudConfig.vcrClusterZkConnectString)) { throw new IllegalArgumentException("Missing value for " + CloudConfig.VCR_CLUSTER_ZK_CONNECT_STRING); } else if (Utils.isNullOrEmpty(cloudConfig.vcrClusterName)) { throw new IllegalArgumentException("Missing value for " + CloudConfig.VCR_CLUSTER_NAME); } this.cloudConfig = cloudConfig; currentDataNode = new CloudDataNode(cloudConfig, clusterMapConfig); List<? extends PartitionId> allPartitions = clusterMap.getAllPartitionIds(null); logger.info("All partitions from clusterMap: {}.", allPartitions); partitionIdMap = allPartitions.stream().collect(Collectors.toMap(PartitionId::toPathString, Function.identity())); vcrClusterName = cloudConfig.vcrClusterName; vcrInstanceName = ClusterMapUtils.getInstanceName(clusterMapConfig.clusterMapHostName, clusterMapConfig.clusterMapPort); metrics = new HelixVcrClusterMetrics(clusterMap.getMetricRegistry(), assignedPartitionIds); } }
HelixVcrCluster implements VirtualReplicatorCluster { public HelixVcrCluster(CloudConfig cloudConfig, ClusterMapConfig clusterMapConfig, ClusterMap clusterMap) { if (Utils.isNullOrEmpty(cloudConfig.vcrClusterZkConnectString)) { throw new IllegalArgumentException("Missing value for " + CloudConfig.VCR_CLUSTER_ZK_CONNECT_STRING); } else if (Utils.isNullOrEmpty(cloudConfig.vcrClusterName)) { throw new IllegalArgumentException("Missing value for " + CloudConfig.VCR_CLUSTER_NAME); } this.cloudConfig = cloudConfig; currentDataNode = new CloudDataNode(cloudConfig, clusterMapConfig); List<? extends PartitionId> allPartitions = clusterMap.getAllPartitionIds(null); logger.info("All partitions from clusterMap: {}.", allPartitions); partitionIdMap = allPartitions.stream().collect(Collectors.toMap(PartitionId::toPathString, Function.identity())); vcrClusterName = cloudConfig.vcrClusterName; vcrInstanceName = ClusterMapUtils.getInstanceName(clusterMapConfig.clusterMapHostName, clusterMapConfig.clusterMapPort); metrics = new HelixVcrClusterMetrics(clusterMap.getMetricRegistry(), assignedPartitionIds); } HelixVcrCluster(CloudConfig cloudConfig, ClusterMapConfig clusterMapConfig, ClusterMap clusterMap); }
HelixVcrCluster implements VirtualReplicatorCluster { public HelixVcrCluster(CloudConfig cloudConfig, ClusterMapConfig clusterMapConfig, ClusterMap clusterMap) { if (Utils.isNullOrEmpty(cloudConfig.vcrClusterZkConnectString)) { throw new IllegalArgumentException("Missing value for " + CloudConfig.VCR_CLUSTER_ZK_CONNECT_STRING); } else if (Utils.isNullOrEmpty(cloudConfig.vcrClusterName)) { throw new IllegalArgumentException("Missing value for " + CloudConfig.VCR_CLUSTER_NAME); } this.cloudConfig = cloudConfig; currentDataNode = new CloudDataNode(cloudConfig, clusterMapConfig); List<? extends PartitionId> allPartitions = clusterMap.getAllPartitionIds(null); logger.info("All partitions from clusterMap: {}.", allPartitions); partitionIdMap = allPartitions.stream().collect(Collectors.toMap(PartitionId::toPathString, Function.identity())); vcrClusterName = cloudConfig.vcrClusterName; vcrInstanceName = ClusterMapUtils.getInstanceName(clusterMapConfig.clusterMapHostName, clusterMapConfig.clusterMapPort); metrics = new HelixVcrClusterMetrics(clusterMap.getMetricRegistry(), assignedPartitionIds); } HelixVcrCluster(CloudConfig cloudConfig, ClusterMapConfig clusterMapConfig, ClusterMap clusterMap); void addPartition(String partitionIdStr); void removePartition(String partitionIdStr); @Override List<? extends DataNodeId> getAllDataNodeIds(); @Override DataNodeId getCurrentDataNodeId(); @Override void participate(); @Override List<? extends PartitionId> getAssignedPartitionIds(); @Override void addListener(VirtualReplicatorClusterListener listener); @Override void close(); }
HelixVcrCluster implements VirtualReplicatorCluster { public HelixVcrCluster(CloudConfig cloudConfig, ClusterMapConfig clusterMapConfig, ClusterMap clusterMap) { if (Utils.isNullOrEmpty(cloudConfig.vcrClusterZkConnectString)) { throw new IllegalArgumentException("Missing value for " + CloudConfig.VCR_CLUSTER_ZK_CONNECT_STRING); } else if (Utils.isNullOrEmpty(cloudConfig.vcrClusterName)) { throw new IllegalArgumentException("Missing value for " + CloudConfig.VCR_CLUSTER_NAME); } this.cloudConfig = cloudConfig; currentDataNode = new CloudDataNode(cloudConfig, clusterMapConfig); List<? extends PartitionId> allPartitions = clusterMap.getAllPartitionIds(null); logger.info("All partitions from clusterMap: {}.", allPartitions); partitionIdMap = allPartitions.stream().collect(Collectors.toMap(PartitionId::toPathString, Function.identity())); vcrClusterName = cloudConfig.vcrClusterName; vcrInstanceName = ClusterMapUtils.getInstanceName(clusterMapConfig.clusterMapHostName, clusterMapConfig.clusterMapPort); metrics = new HelixVcrClusterMetrics(clusterMap.getMetricRegistry(), assignedPartitionIds); } HelixVcrCluster(CloudConfig cloudConfig, ClusterMapConfig clusterMapConfig, ClusterMap clusterMap); void addPartition(String partitionIdStr); void removePartition(String partitionIdStr); @Override List<? extends DataNodeId> getAllDataNodeIds(); @Override DataNodeId getCurrentDataNodeId(); @Override void participate(); @Override List<? extends PartitionId> getAssignedPartitionIds(); @Override void addListener(VirtualReplicatorClusterListener listener); @Override void close();  }