@Test public void testExecute() throws ExecuteException { UpdateSnapshotCacheForQueryServersStep step = new UpdateSnapshotCacheForQueryServersStep(); ExecuteResult result = step.doWork(new DefaultContext(Maps.<String, Executable>newConcurrentMap(), kylinConfig)); System.out.println(result.output()); assertTrue(result.succeed()); }
@Override protected ExecuteResult doWork(ExecutableContext context) throws ExecuteException { final String tableName = LookupExecutableUtil.getLookupTableName(this.getParams()); final String snapshotID = LookupExecutableUtil.getLookupSnapshotID(this.getParams()); final String projectName = LookupExecutableUtil.getProjectName(this.getParams()); final KylinConfig config = KylinConfig.getInstanceFromEnv(); int checkInterval = 10 * 1000; int maxCheckTime = 10 * 60 * 1000; StringWriter outputWriter = new StringWriter(); PrintWriter pw = new PrintWriter(outputWriter); String[] restServers = config.getRestServers(); List<String> serversNeedCheck = Lists.newArrayList(); for (String restServer : restServers) { logger.info("send build lookup table cache request to server: " + restServer); try { RestClient restClient = new RestClient(restServer); restClient.buildLookupSnapshotCache(projectName, tableName, snapshotID); serversNeedCheck.add(restServer); } catch (IOException e) { logger.error("error when send build cache request to rest server:" + restServer, e); pw.println("cache build fail for rest server:" + restServer); } } if (serversNeedCheck.isEmpty()) { return new ExecuteResult(State.SUCCEED, outputWriter.toString()); } List<String> completeServers = Lists.newArrayList(); long startTime = System.currentTimeMillis(); while ((System.currentTimeMillis() - startTime) < maxCheckTime) { serversNeedCheck.removeAll(completeServers); if (serversNeedCheck.isEmpty()) { break; } for (String restServer : serversNeedCheck) { logger.info("check lookup table cache build status for server: " + restServer); try { RestClient restClient = new RestClient(restServer); String stateName = restClient.getLookupSnapshotCacheState(tableName, snapshotID); if (!stateName.equals(CacheState.IN_BUILDING.name())) { completeServers.add(restServer); pw.println("cache build complete for rest server:" + restServer + " cache state:" + stateName); } } catch (IOException e) { logger.error("error when send build cache request to rest server:" + restServer, e); } } try { Thread.sleep(checkInterval); } catch (InterruptedException e) { logger.error("interrupted", e); } } serversNeedCheck.removeAll(completeServers); if (!serversNeedCheck.isEmpty()) { pw.println(); pw.println("check timeout!"); pw.println("servers not complete:" + serversNeedCheck); } return new ExecuteResult(State.SUCCEED, outputWriter.toString()); }
UpdateSnapshotCacheForQueryServersStep extends AbstractExecutable { @Override protected ExecuteResult doWork(ExecutableContext context) throws ExecuteException { final String tableName = LookupExecutableUtil.getLookupTableName(this.getParams()); final String snapshotID = LookupExecutableUtil.getLookupSnapshotID(this.getParams()); final String projectName = LookupExecutableUtil.getProjectName(this.getParams()); final KylinConfig config = KylinConfig.getInstanceFromEnv(); int checkInterval = 10 * 1000; int maxCheckTime = 10 * 60 * 1000; StringWriter outputWriter = new StringWriter(); PrintWriter pw = new PrintWriter(outputWriter); String[] restServers = config.getRestServers(); List<String> serversNeedCheck = Lists.newArrayList(); for (String restServer : restServers) { logger.info("send build lookup table cache request to server: " + restServer); try { RestClient restClient = new RestClient(restServer); restClient.buildLookupSnapshotCache(projectName, tableName, snapshotID); serversNeedCheck.add(restServer); } catch (IOException e) { logger.error("error when send build cache request to rest server:" + restServer, e); pw.println("cache build fail for rest server:" + restServer); } } if (serversNeedCheck.isEmpty()) { return new ExecuteResult(State.SUCCEED, outputWriter.toString()); } List<String> completeServers = Lists.newArrayList(); long startTime = System.currentTimeMillis(); while ((System.currentTimeMillis() - startTime) < maxCheckTime) { serversNeedCheck.removeAll(completeServers); if (serversNeedCheck.isEmpty()) { break; } for (String restServer : serversNeedCheck) { logger.info("check lookup table cache build status for server: " + restServer); try { RestClient restClient = new RestClient(restServer); String stateName = restClient.getLookupSnapshotCacheState(tableName, snapshotID); if (!stateName.equals(CacheState.IN_BUILDING.name())) { completeServers.add(restServer); pw.println("cache build complete for rest server:" + restServer + " cache state:" + stateName); } } catch (IOException e) { logger.error("error when send build cache request to rest server:" + restServer, e); } } try { Thread.sleep(checkInterval); } catch (InterruptedException e) { logger.error("interrupted", e); } } serversNeedCheck.removeAll(completeServers); if (!serversNeedCheck.isEmpty()) { pw.println(); pw.println("check timeout!"); pw.println("servers not complete:" + serversNeedCheck); } return new ExecuteResult(State.SUCCEED, outputWriter.toString()); } }
UpdateSnapshotCacheForQueryServersStep extends AbstractExecutable { @Override protected ExecuteResult doWork(ExecutableContext context) throws ExecuteException { final String tableName = LookupExecutableUtil.getLookupTableName(this.getParams()); final String snapshotID = LookupExecutableUtil.getLookupSnapshotID(this.getParams()); final String projectName = LookupExecutableUtil.getProjectName(this.getParams()); final KylinConfig config = KylinConfig.getInstanceFromEnv(); int checkInterval = 10 * 1000; int maxCheckTime = 10 * 60 * 1000; StringWriter outputWriter = new StringWriter(); PrintWriter pw = new PrintWriter(outputWriter); String[] restServers = config.getRestServers(); List<String> serversNeedCheck = Lists.newArrayList(); for (String restServer : restServers) { logger.info("send build lookup table cache request to server: " + restServer); try { RestClient restClient = new RestClient(restServer); restClient.buildLookupSnapshotCache(projectName, tableName, snapshotID); serversNeedCheck.add(restServer); } catch (IOException e) { logger.error("error when send build cache request to rest server:" + restServer, e); pw.println("cache build fail for rest server:" + restServer); } } if (serversNeedCheck.isEmpty()) { return new ExecuteResult(State.SUCCEED, outputWriter.toString()); } List<String> completeServers = Lists.newArrayList(); long startTime = System.currentTimeMillis(); while ((System.currentTimeMillis() - startTime) < maxCheckTime) { serversNeedCheck.removeAll(completeServers); if (serversNeedCheck.isEmpty()) { break; } for (String restServer : serversNeedCheck) { logger.info("check lookup table cache build status for server: " + restServer); try { RestClient restClient = new RestClient(restServer); String stateName = restClient.getLookupSnapshotCacheState(tableName, snapshotID); if (!stateName.equals(CacheState.IN_BUILDING.name())) { completeServers.add(restServer); pw.println("cache build complete for rest server:" + restServer + " cache state:" + stateName); } } catch (IOException e) { logger.error("error when send build cache request to rest server:" + restServer, e); } } try { Thread.sleep(checkInterval); } catch (InterruptedException e) { logger.error("interrupted", e); } } serversNeedCheck.removeAll(completeServers); if (!serversNeedCheck.isEmpty()) { pw.println(); pw.println("check timeout!"); pw.println("servers not complete:" + serversNeedCheck); } return new ExecuteResult(State.SUCCEED, outputWriter.toString()); }  }
UpdateSnapshotCacheForQueryServersStep extends AbstractExecutable { @Override protected ExecuteResult doWork(ExecutableContext context) throws ExecuteException { final String tableName = LookupExecutableUtil.getLookupTableName(this.getParams()); final String snapshotID = LookupExecutableUtil.getLookupSnapshotID(this.getParams()); final String projectName = LookupExecutableUtil.getProjectName(this.getParams()); final KylinConfig config = KylinConfig.getInstanceFromEnv(); int checkInterval = 10 * 1000; int maxCheckTime = 10 * 60 * 1000; StringWriter outputWriter = new StringWriter(); PrintWriter pw = new PrintWriter(outputWriter); String[] restServers = config.getRestServers(); List<String> serversNeedCheck = Lists.newArrayList(); for (String restServer : restServers) { logger.info("send build lookup table cache request to server: " + restServer); try { RestClient restClient = new RestClient(restServer); restClient.buildLookupSnapshotCache(projectName, tableName, snapshotID); serversNeedCheck.add(restServer); } catch (IOException e) { logger.error("error when send build cache request to rest server:" + restServer, e); pw.println("cache build fail for rest server:" + restServer); } } if (serversNeedCheck.isEmpty()) { return new ExecuteResult(State.SUCCEED, outputWriter.toString()); } List<String> completeServers = Lists.newArrayList(); long startTime = System.currentTimeMillis(); while ((System.currentTimeMillis() - startTime) < maxCheckTime) { serversNeedCheck.removeAll(completeServers); if (serversNeedCheck.isEmpty()) { break; } for (String restServer : serversNeedCheck) { logger.info("check lookup table cache build status for server: " + restServer); try { RestClient restClient = new RestClient(restServer); String stateName = restClient.getLookupSnapshotCacheState(tableName, snapshotID); if (!stateName.equals(CacheState.IN_BUILDING.name())) { completeServers.add(restServer); pw.println("cache build complete for rest server:" + restServer + " cache state:" + stateName); } } catch (IOException e) { logger.error("error when send build cache request to rest server:" + restServer, e); } } try { Thread.sleep(checkInterval); } catch (InterruptedException e) { logger.error("interrupted", e); } } serversNeedCheck.removeAll(completeServers); if (!serversNeedCheck.isEmpty()) { pw.println(); pw.println("check timeout!"); pw.println("servers not complete:" + serversNeedCheck); } return new ExecuteResult(State.SUCCEED, outputWriter.toString()); }   }
UpdateSnapshotCacheForQueryServersStep extends AbstractExecutable { @Override protected ExecuteResult doWork(ExecutableContext context) throws ExecuteException { final String tableName = LookupExecutableUtil.getLookupTableName(this.getParams()); final String snapshotID = LookupExecutableUtil.getLookupSnapshotID(this.getParams()); final String projectName = LookupExecutableUtil.getProjectName(this.getParams()); final KylinConfig config = KylinConfig.getInstanceFromEnv(); int checkInterval = 10 * 1000; int maxCheckTime = 10 * 60 * 1000; StringWriter outputWriter = new StringWriter(); PrintWriter pw = new PrintWriter(outputWriter); String[] restServers = config.getRestServers(); List<String> serversNeedCheck = Lists.newArrayList(); for (String restServer : restServers) { logger.info("send build lookup table cache request to server: " + restServer); try { RestClient restClient = new RestClient(restServer); restClient.buildLookupSnapshotCache(projectName, tableName, snapshotID); serversNeedCheck.add(restServer); } catch (IOException e) { logger.error("error when send build cache request to rest server:" + restServer, e); pw.println("cache build fail for rest server:" + restServer); } } if (serversNeedCheck.isEmpty()) { return new ExecuteResult(State.SUCCEED, outputWriter.toString()); } List<String> completeServers = Lists.newArrayList(); long startTime = System.currentTimeMillis(); while ((System.currentTimeMillis() - startTime) < maxCheckTime) { serversNeedCheck.removeAll(completeServers); if (serversNeedCheck.isEmpty()) { break; } for (String restServer : serversNeedCheck) { logger.info("check lookup table cache build status for server: " + restServer); try { RestClient restClient = new RestClient(restServer); String stateName = restClient.getLookupSnapshotCacheState(tableName, snapshotID); if (!stateName.equals(CacheState.IN_BUILDING.name())) { completeServers.add(restServer); pw.println("cache build complete for rest server:" + restServer + " cache state:" + stateName); } } catch (IOException e) { logger.error("error when send build cache request to rest server:" + restServer, e); } } try { Thread.sleep(checkInterval); } catch (InterruptedException e) { logger.error("interrupted", e); } } serversNeedCheck.removeAll(completeServers); if (!serversNeedCheck.isEmpty()) { pw.println(); pw.println("check timeout!"); pw.println("servers not complete:" + serversNeedCheck); } return new ExecuteResult(State.SUCCEED, outputWriter.toString()); }    }