HTTPCallbackAction extends PostAction { public HTTPCallbackAction(HttpRequestConf conf) { this.conf = conf; } HTTPCallbackAction(HttpRequestConf conf); @Override void configurePipeline(PipelineConfigurer pipelineConfigurer); @SuppressWarnings("ConstantConditions") @Override void run(BatchActionContext batchActionContext);  }
@Test public void testHTTPCallbackAction() throws Exception { String body = "samuel jackson, dwayne johnson, christopher walken"; ETLStage action = new ETLStage( "http", new ETLPlugin("HTTPCallback", PostAction.PLUGIN_TYPE, ImmutableMap.of("url", "${url}", "method", "${method}", "body", "${body}"), null)); ETLStage source = new ETLStage("source", MockSource.getPlugin("httpCallbackInput")); ETLStage sink = new ETLStage("sink", MockSink.getPlugin("httpCallbackOutput")); ETLBatchConfig etlConfig = ETLBatchConfig.builder("* * * * *") .addStage(source) .addStage(sink) .addPostAction(action) .addConnection(source.getName(), sink.getName()) .build(); AppRequest<ETLBatchConfig> appRequest = new AppRequest<>(BATCH_ARTIFACT, etlConfig); ApplicationId appId = NamespaceId.DEFAULT.app("actionTest"); ApplicationManager appManager = TestBase.deployApplication(appId, appRequest); Map<String, String> arguments = new HashMap<>(); arguments.put("url", baseURL + "/feeds/users"); arguments.put("method", "PUT"); arguments.put("body", body); WorkflowManager manager = appManager.getWorkflowManager(SmartWorkflow.NAME); manager.start(arguments); manager.waitForRuns(ProgramRunStatus.COMPLETED, 1, 5, TimeUnit.MINUTES); Assert.assertEquals(body, getFeedContent("users")); }