Repository: incubator-gobblin Updated Branches: refs/heads/master f121bb2c8 -> 7526439ca
[GOBBLIN-601] Cancel a flow by AzkabanClient Closes #2467 from yukuai518/azkCancel Project: http://git-wip-us.apache.org/repos/asf/incubator-gobblin/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-gobblin/commit/7526439c Tree: http://git-wip-us.apache.org/repos/asf/incubator-gobblin/tree/7526439c Diff: http://git-wip-us.apache.org/repos/asf/incubator-gobblin/diff/7526439c Branch: refs/heads/master Commit: 7526439ca635da0973646b5621bbce35107f69e6 Parents: f121bb2 Author: Kuai Yu <[email protected]> Authored: Mon Oct 8 09:29:30 2018 -0700 Committer: Hung Tran <[email protected]> Committed: Mon Oct 8 09:29:30 2018 -0700 ---------------------------------------------------------------------- .../modules/orchestration/AzkabanClient.java | 30 ++++++++++++++++++++ 1 file changed, 30 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-gobblin/blob/7526439c/gobblin-modules/gobblin-azkaban/src/main/java/org/apache/gobblin/service/modules/orchestration/AzkabanClient.java ---------------------------------------------------------------------- diff --git a/gobblin-modules/gobblin-azkaban/src/main/java/org/apache/gobblin/service/modules/orchestration/AzkabanClient.java b/gobblin-modules/gobblin-azkaban/src/main/java/org/apache/gobblin/service/modules/orchestration/AzkabanClient.java index 624e24a..cccbb5e 100644 --- a/gobblin-modules/gobblin-azkaban/src/main/java/org/apache/gobblin/service/modules/orchestration/AzkabanClient.java +++ b/gobblin-modules/gobblin-azkaban/src/main/java/org/apache/gobblin/service/modules/orchestration/AzkabanClient.java @@ -421,6 +421,36 @@ public class AzkabanClient implements Closeable { } /** + * Cancel a flow by execution id. + */ + public AzkabanClientStatus cancelFlow(int execId) { + try { + refreshSession(); + List<NameValuePair> nvps = new ArrayList<>(); + nvps.add(new BasicNameValuePair(AzkabanClientParams.AJAX, "cancelFlow")); + nvps.add(new BasicNameValuePair(AzkabanClientParams.SESSION_ID, this.sessionId)); + nvps.add(new BasicNameValuePair(AzkabanClientParams.EXECID, String.valueOf(execId))); + + Header contentType = new BasicHeader(HttpHeaders.CONTENT_TYPE, "application/x-www-form-urlencoded"); + Header requestType = new BasicHeader("X-Requested-With", "XMLHttpRequest"); + + HttpGet httpGet = new HttpGet(url + "/executor?" + URLEncodedUtils.format(nvps, "UTF-8")); + httpGet.setHeaders(new Header[]{contentType, requestType}); + + CloseableHttpResponse response = this.httpClient.execute(httpGet); + try { + handleResponse(response); + return new AzkabanClientStatus.SUCCESS(); + } finally { + response.close(); + } + } catch (Exception e) { + return new AzkabanClientStatus.FAIL("", e); + } + } + + + /** * Given an execution id, fetches all the detailed information of that execution, including a list of all the job executions. * * @param execId execution id to be fetched.
