@Test public void testReplicaAssignmentChangeEffects() throws Exception { IntegrationTestHelper helper = null; ZooKeeperClient zkClient = null; try { String rootPath = "/storeimpl/test"; Properties properties = new Properties(); properties.setProperty(IntegrationTestHelper.Config.ZNODE_PATH, rootPath); properties.setProperty(IntegrationTestHelper.Config.NUM_PARTITIONS, String.valueOf(CLUSTER_NUM_PARTITIONS)); properties.setProperty(IntegrationTestHelper.Config.ZK_SESSION_TIMEOUT, "30000"); properties.setProperty(IntegrationTestHelper.Config.NUM_STORAGES, String.valueOf(NUM_REPLICAS)); helper = new IntegrationTestHelper(properties); helper.startZooKeeperServer(); List<String> storageConnectStrings = new ArrayList<>(); List<Integer> adminPorts = new ArrayList<>(); for (int s = 0; s < NUM_REPLICAS; s++) { WaltzStorageRunner storageRunner = helper.getWaltzStorageRunner(s); storageRunner.startAsync(); WaltzStorage storage = storageRunner.awaitStart(); adminPorts.add(storage.adminPort); storageConnectStrings.add(helper.getStorageConnectString(s)); } Thread.sleep(500); for (int adminPort : adminPorts) { helper.setWaltzStorageAssignmentWithPort(adminPort, true); } helper.startWaltzServer(true); Thread.sleep(500); ZNode root = new ZNode(rootPath); zkClient = new ZooKeeperClientImpl(helper.getZkConnectString(), helper.getZkSessionTimeout()); Map<Integer, Long> partitionSessionIds = new HashMap<>(); Map<Integer, List<String>> partitionReplicas = new HashMap<>(); populateReplicaStateInfo(zkClient, root, partitionSessionIds, partitionReplicas); assertEquals(CLUSTER_NUM_PARTITIONS, partitionSessionIds.size()); assertEquals(CLUSTER_NUM_PARTITIONS, partitionReplicas.size()); partitionReplicas.forEach((partitionId, replicaConnectStrings) -> assertEquals(new HashSet<>(storageConnectStrings), new HashSet<>(replicaConnectStrings)) ); StoreMetadata storeMetadata = new StoreMetadata(zkClient, new ZNode(root, StoreMetadata.STORE_ZNODE_NAME)); for (int partitionId = 0; partitionId < CLUSTER_NUM_PARTITIONS; partitionId++) { storeMetadata.removePartition(partitionId, helper.getStorageConnectString(0)); } Thread.sleep(1000); Map<Integer, Long> newPartitionSessionIds = new HashMap<>(); Map<Integer, List<String>> newPartitionReplicas = new HashMap<>(); populateReplicaStateInfo(zkClient, root, newPartitionSessionIds, newPartitionReplicas); assertEquals(CLUSTER_NUM_PARTITIONS, partitionSessionIds.size()); assertEquals(CLUSTER_NUM_PARTITIONS, partitionReplicas.size()); newPartitionReplicas.forEach((partitionId, replicaConnectStrings) -> { assertEquals(NUM_REPLICAS - 1, replicaConnectStrings.size()); assertTrue(storageConnectStrings.containsAll(replicaConnectStrings)); }); newPartitionSessionIds.forEach((partitionId, sessionId) -> assertTrue(sessionId > partitionSessionIds.get(partitionId)) ); } finally { if (zkClient != null) { zkClient.close(); } if (helper != null) { helper.closeAll(); } } }
@Override public void close() { if (running.compareAndSet(true, false)) { assignmentWatcherHandle.close(); replicaSessionManager.close(); storeSessionManagers.clear(); asyncTaskExecutor.shutdownNow(); } }
StoreImpl implements Store { @Override public void close() { if (running.compareAndSet(true, false)) { assignmentWatcherHandle.close(); replicaSessionManager.close(); storeSessionManagers.clear(); asyncTaskExecutor.shutdownNow(); } } }
StoreImpl implements Store { @Override public void close() { if (running.compareAndSet(true, false)) { assignmentWatcherHandle.close(); replicaSessionManager.close(); storeSessionManagers.clear(); asyncTaskExecutor.shutdownNow(); } } StoreImpl(ZooKeeperClient zkClient, ZNode storeRoot, WaltzServerConfig config); }
StoreImpl implements Store { @Override public void close() { if (running.compareAndSet(true, false)) { assignmentWatcherHandle.close(); replicaSessionManager.close(); storeSessionManagers.clear(); asyncTaskExecutor.shutdownNow(); } } StoreImpl(ZooKeeperClient zkClient, ZNode storeRoot, WaltzServerConfig config); @Override void close(); @Override StorePartition getPartition(int partitionId, int generation); Set<ReplicaId> getReplicaIds(); ConnectionConfig getConnectionConfig(); void onPartitionRemoved(int partitionId); }
StoreImpl implements Store { @Override public void close() { if (running.compareAndSet(true, false)) { assignmentWatcherHandle.close(); replicaSessionManager.close(); storeSessionManagers.clear(); asyncTaskExecutor.shutdownNow(); } } StoreImpl(ZooKeeperClient zkClient, ZNode storeRoot, WaltzServerConfig config); @Override void close(); @Override StorePartition getPartition(int partitionId, int generation); Set<ReplicaId> getReplicaIds(); ConnectionConfig getConnectionConfig(); void onPartitionRemoved(int partitionId);  }