@SuppressWarnings("unchecked") @Test public void testSyncPartitionAssignments() throws Exception { int numStorage = 3; int numPartition = 12; int zkSessionTimeout = 30000; int zkConnectTimeout = 10000; int storageGroupId = 0; long segmentSizeThreshold = 400L; String localhost = InetAddress.getLocalHost().getCanonicalHostName(); PortFinder portFinder = new PortFinder(); String znodePath = "/storage/cli/test"; ZNode clusterRoot = new ZNode(znodePath); ZNode storeRoot = new ZNode(clusterRoot, StoreMetadata.STORE_ZNODE_NAME); SslSetup sslSetup = new SslSetup(); Path sslDir = Files.createTempDirectory("waltz-integration-test"); String sslConfigPath = IntegrationTestHelper.createYamlConfigFile(sslDir, "ssl.yaml", WaltzStorageConfig.STORAGE_SSL_CONFIG_PREFIX, sslSetup); SslContext sslContext = Utils.getSslContext(sslConfigPath, WaltzStorageConfig.STORAGE_SSL_CONFIG_PREFIX); ZooKeeperServerRunner zkServerRunner = null; ZooKeeperClient zkClient = null; List<WaltzStorageRunner> storageRunners = new ArrayList<>(); Map<String, StorageClient> storageClientMap = new HashMap<>(); Map<String, StorageAdminClient> storageAdminClientMap = new HashMap<>(); try { zkServerRunner = new ZooKeeperServerRunner(portFinder.getPort()); String zkConnectionString = zkServerRunner.start(); zkClient = new ZooKeeperClientImpl(zkConnectionString, zkSessionTimeout, zkConnectTimeout); ZooKeeperCli.Create.createCluster(zkClient, clusterRoot, "test cluster", numPartition); ZooKeeperCli.Create.createStores(zkClient, clusterRoot, numPartition); StoreMetadata storeMetadata = new StoreMetadata(zkClient, storeRoot); UUID key = storeMetadata.getStoreParams().key; for (int i = 0; i < numStorage; i++) { Properties storageProperties = new Properties(); storageProperties.setProperty(WaltzStorageConfig.STORAGE_JETTY_PORT, String.valueOf(portFinder.getPort())); storageProperties.setProperty(WaltzStorageConfig.STORAGE_DIRECTORY, Files.createTempDirectory(DIR_NAME).toString()); storageProperties.setProperty(WaltzStorageConfig.SEGMENT_SIZE_THRESHOLD, String.valueOf(segmentSizeThreshold)); storageProperties.setProperty(WaltzStorageConfig.ZOOKEEPER_CONNECT_STRING, zkConnectionString); storageProperties.setProperty(WaltzStorageConfig.ZOOKEEPER_SESSION_TIMEOUT, String.valueOf(zkSessionTimeout)); storageProperties.setProperty(WaltzStorageConfig.CLUSTER_ROOT, clusterRoot.path); sslSetup.setConfigParams(storageProperties, WaltzStorageConfig.STORAGE_SSL_CONFIG_PREFIX); WaltzStorageConfig storageConfig = new WaltzStorageConfig(storageProperties); WaltzStorageRunner storageRunner = new WaltzStorageRunner(portFinder, storageConfig, segmentSizeThreshold); storageRunner.startAsync(); WaltzStorage storage = storageRunner.awaitStart(); String connectString = localhost + ":" + storage.port; StorageClient storageClient = new StorageClient(localhost, storage.port, sslContext, key, numPartition); storageClient.open(); StorageAdminClient storageAdminClient = new StorageAdminClient(localhost, storage.adminPort, sslContext, key, numPartition); storageAdminClient.open(); storeMetadata.addStorageNode(connectString, storageGroupId, storage.adminPort); storageClientMap.put(localhost + ":" + storage.port, storageClient); storageAdminClientMap.put(localhost + ":" + storage.port, storageAdminClient); storageRunners.add(storageRunner); } storeMetadata.autoAssignPartition(storageGroupId); for (Map.Entry<String, int[]> entry : storeMetadata.getReplicaAssignments().replicas.entrySet()) { String connectString = entry.getKey(); int[] partitionIds = entry.getValue(); assertEquals(((Map<String, Boolean>) storageAdminClientMap.get(connectString).getAssignedPartitionStatus().get()).size(), 0); for (int partitionId : partitionIds) { try { storageClientMap.get(connectString).lastSessionInfo(0L, partitionId).get(); fail("Unassigned storage client should fail to get last session info."); } catch (ExecutionException e) { Throwable t = e.getCause(); assertTrue(t.getMessage().contains("Partition:" + partitionId + " is not assigned.")); } } } Properties cfgProperties = createProperties(zkConnectionString, znodePath, sslSetup); String configFilePath = IntegrationTestHelper.createYamlConfigFile(DIR_NAME, CONFIG_FILE_NAME, cfgProperties); String[] args = { "sync-partitions", "--cli-config-path", configFilePath }; StorageCli.testMain(args); for (Map.Entry<String, int[]> entry : storeMetadata.getReplicaAssignments().replicas.entrySet()) { assertEquals(((Map<String, Boolean>) storageAdminClientMap.get(entry.getKey()).getAssignedPartitionStatus().get()).size(), entry.getValue().length); for (int partitionId : entry.getValue()) { storageClientMap.get(entry.getKey()).lastSessionInfo(0L, partitionId).get(); } } } finally { for (StorageClient storageClient : storageClientMap.values()) { storageClient.close(); } for (StorageAdminClient storageAdminClient : storageAdminClientMap.values()) { storageAdminClient.close(); } for (WaltzStorageRunner runner : storageRunners) { runner.stop(); } if (zkClient != null) { try { zkClient.deleteRecursively(clusterRoot); } finally { zkClient.close(); } } if (zkServerRunner != null) { zkServerRunner.stop(); } sslSetup.close(); Utils.removeDirectory(sslDir.toFile()); } }
public static void testMain(String[] args) { new StorageCli(args, true).processCmd(); }
StorageCli extends SubcommandCli { public static void testMain(String[] args) { new StorageCli(args, true).processCmd(); } }
StorageCli extends SubcommandCli { public static void testMain(String[] args) { new StorageCli(args, true).processCmd(); } private  StorageCli(String[] args, boolean useByTest); }
StorageCli extends SubcommandCli { public static void testMain(String[] args) { new StorageCli(args, true).processCmd(); } private  StorageCli(String[] args, boolean useByTest); static void testMain(String[] args); static void main(String[] args); }
StorageCli extends SubcommandCli { public static void testMain(String[] args) { new StorageCli(args, true).processCmd(); } private  StorageCli(String[] args, boolean useByTest); static void testMain(String[] args); static void main(String[] args);  }