@Test public void testRecoverPartition() throws Exception { long sessionId = 123; int numPartitions = 3; IntegrationTestHelper helper = getIntegrationTestHelper(numPartitions); String host = helper.getHost(); SslContext sslCtx = helper.getSslContext(); int partitionId = new Random().nextInt(helper.getNumPartitions()); Properties cfgProperties = createProperties(helper.getZkConnectString(), helper.getZnodePath(), helper.getSslSetup()); Properties sslCfgProperties = createProperties(helper.getZkConnectString(), helper.getZnodePath(), helper.getSslSetup(), WaltzServerConfig.SERVER_SSL_CONFIG_PREFIX); String cliConfigPath = IntegrationTestHelper.createYamlConfigFile(DIR_NAME, CONFIG_FILE_NAME, cfgProperties); String sourceSslConfigPath = IntegrationTestHelper.createYamlConfigFile(DIR_NAME, SOURCE_SSL_CONFIG_FILE_NAME, sslCfgProperties); String destinationSslConfigPath = IntegrationTestHelper.createYamlConfigFile(DIR_NAME, DESTINATION_SSL_CONFIG_FILE_NAME, sslCfgProperties); helper.startZooKeeperServer(); int destinationPort = helper.getPort(); int destinationAdminPort = helper.getPort(); int destinationJettyPort = helper.getPort(); WaltzStorageRunner destinationStorageRunner = helper.getWaltzStorageRunner(destinationPort, destinationAdminPort, destinationJettyPort); WaltzStorageRunner sourceStorageRunner = helper.getWaltzStorageRunner(); UUID key = helper.getClusterKey(); try { sourceStorageRunner.startAsync(); destinationStorageRunner.startAsync(); WaltzStorage sourceWaltzStorage = sourceStorageRunner.awaitStart(); WaltzStorage destinationWaltzStorage = destinationStorageRunner.awaitStart(); int sourceAdminPort = sourceWaltzStorage.adminPort; for (WaltzStorage waltzStorage : Utils.list(sourceWaltzStorage, destinationWaltzStorage)) { int adminPort = waltzStorage.adminPort; StorageAdminClient adminClient = helper.getStorageAdminClientWithAdminPort(adminPort); adminClient.open(); for (int i = 0; i < numPartitions; i++) { adminClient.setPartitionAssignment(i, true, false).get(); } adminClient.close(); } ArrayList<Record> records = ClientUtil.makeRecords(0, 10); int sourcePort = sourceWaltzStorage.port; StorageClient sourceClient = new StorageClient(host, sourcePort, sslCtx, key, numPartitions); sourceClient.open(); sourceClient.setLowWaterMark(sessionId, partitionId, -1L).get(); sourceClient.appendRecords(sessionId, partitionId, records).get(); sourceClient.setLowWaterMark(sessionId + 1, partitionId, records.size() - 1).get(); StorageClient destinationClient = new StorageClient(host, destinationPort, sslCtx, key, numPartitions); destinationClient.open(); Long maxTransactionId = (Long) destinationClient.getMaxTransactionId(-1, partitionId).get(); assertNotNull(maxTransactionId); assertEquals(-1, maxTransactionId.longValue()); String[] args0 = { "availability", "--storage", host + ":" + destinationAdminPort, "--partition", String.valueOf(partitionId), "--online", String.valueOf(false), "--cli-config-path", cliConfigPath }; StorageCli.testMain(args0); String[] args1 = { "recover-partition", "--source-storage", host + ":" + sourceAdminPort, "--destination-storage", host + ":" + destinationAdminPort, "--destination-storage-port", String.valueOf(destinationPort), "--partition", String.valueOf(partitionId), "--batch-size", String.valueOf(20), "--cli-config-path", cliConfigPath, "--source-ssl-config-path", sourceSslConfigPath, "--destination-ssl-config-path", destinationSslConfigPath, }; StorageCli.testMain(args1); String[] args2 = { "availability", "--storage", host + ":" + destinationAdminPort, "--partition", String.valueOf(partitionId), "--online", String.valueOf(true), "--cli-config-path", cliConfigPath }; StorageCli.testMain(args2); maxTransactionId = (Long) destinationClient.getMaxTransactionId(-1, partitionId).get(); assertNotNull(maxTransactionId); assertEquals(9, maxTransactionId.longValue()); sourceClient.close(); destinationClient.close(); } finally { helper.closeAll(); } }
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);  }