You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ambari.apache.org by ni...@apache.org on 2016/12/28 09:33:19 UTC
[19/23] ambari git commit: AMBARI-19302 : removed contrib/views/hive
folder and made necessary changes in pom.xml files (nitirajrathore)
http://git-wip-us.apache.org/repos/asf/ambari/blob/c0f9621f/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/ProgressRetriever.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/ProgressRetriever.java b/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/ProgressRetriever.java
deleted file mode 100644
index 6be8147..0000000
--- a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/ProgressRetriever.java
+++ /dev/null
@@ -1,67 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ambari.view.hive.resources.jobs;
-
-import org.apache.ambari.view.hive.resources.jobs.atsJobs.TezVertexId;
-import org.apache.ambari.view.hive.resources.jobs.rm.RMParser;
-import org.apache.ambari.view.hive.resources.jobs.viewJobs.Job;
-import org.apache.ambari.view.hive.utils.ServiceFormattedException;
-import org.apache.ambari.view.hive.utils.SharedObjectsFactory;
-
-import java.util.List;
-import java.util.Map;
-
-public class ProgressRetriever {
- private final Progress progress;
- private final Job job;
- private final SharedObjectsFactory sharedObjects;
-
- public ProgressRetriever(Job job, SharedObjectsFactory sharedObjects) {
- this.job = job;
- this.sharedObjects = sharedObjects;
-
- this.progress = new Progress();
- }
-
- public Progress getProgress() {
- jobCheck();
-
- progress.dagProgress = sharedObjects.getRMParser().getDAGProgress(
- job.getApplicationId(), job.getDagId());
-
- List<TezVertexId> vertices = sharedObjects.getATSParser().getVerticesForDAGId(job.getDagId());
- progress.vertexProgresses = sharedObjects.getRMParser().getDAGVerticesProgress(job.getApplicationId(), job.getDagId(), vertices);
-
- return progress;
- }
-
- public void jobCheck() {
- if (job.getApplicationId() == null || job.getApplicationId().isEmpty()) {
- throw new ServiceFormattedException("E070 ApplicationId is not defined yet");
- }
- if (job.getDagId() == null || job.getDagId().isEmpty()) {
- throw new ServiceFormattedException("E080 DagID is not defined yet");
- }
- }
-
- public static class Progress {
- public Double dagProgress;
- public List<RMParser.VertexProgress> vertexProgresses;
- }
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c0f9621f/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/ResultsPaginationController.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/ResultsPaginationController.java b/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/ResultsPaginationController.java
deleted file mode 100644
index cc2ff42..0000000
--- a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/ResultsPaginationController.java
+++ /dev/null
@@ -1,240 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ambari.view.hive.resources.jobs;
-
-
-import org.apache.ambari.view.ViewContext;
-import org.apache.ambari.view.hive.client.ColumnDescription;
-import org.apache.ambari.view.hive.client.HiveClientException;
-import org.apache.ambari.view.hive.client.Cursor;
-import org.apache.ambari.view.hive.utils.HiveClientFormattedException;
-import org.apache.ambari.view.hive.utils.ServiceFormattedException;
-import org.apache.commons.collections4.map.PassiveExpiringMap;
-
-import javax.ws.rs.core.Response;
-import java.lang.Object;
-import java.lang.String;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.List;
-import java.util.concurrent.Callable;
-
-/**
- * Results Pagination Controller
- * Persists cursors for result sets
- */
-public class ResultsPaginationController {
- public static final String DEFAULT_SEARCH_ID = "default";
- private static Map<String, ResultsPaginationController> viewSingletonObjects = new HashMap<String, ResultsPaginationController>();
- public static ResultsPaginationController getInstance(ViewContext context) {
- if (!viewSingletonObjects.containsKey(context.getInstanceName()))
- viewSingletonObjects.put(context.getInstanceName(), new ResultsPaginationController());
- return viewSingletonObjects.get(context.getInstanceName());
- }
-
- public ResultsPaginationController() {
- }
-
- private static final long EXPIRING_TIME = 10*60*1000; // 10 minutes
- private static final int DEFAULT_FETCH_COUNT = 50;
- private Map<String, Cursor> resultsCache;
-
- public static class CustomTimeToLiveExpirationPolicy extends PassiveExpiringMap.ConstantTimeToLiveExpirationPolicy<String, Cursor> {
- public CustomTimeToLiveExpirationPolicy(long timeToLiveMillis) {
- super(timeToLiveMillis);
- }
-
- @Override
- public long expirationTime(String key, Cursor value) {
- if (key.startsWith("$")) {
- return -1; //never expire
- }
- return super.expirationTime(key, value);
- }
- }
-
- private Map<String, Cursor> getResultsCache() {
- if (resultsCache == null) {
- PassiveExpiringMap<String, Cursor> resultsCacheExpiringMap =
- new PassiveExpiringMap<String, Cursor>(new CustomTimeToLiveExpirationPolicy(EXPIRING_TIME));
- resultsCache = Collections.synchronizedMap(resultsCacheExpiringMap);
- }
- return resultsCache;
- }
-
- /**
- * Renew timer of cache entry.
- * @param key name/id of results request
- * @return false if entry not found; true if renew was ok
- */
- public boolean keepAlive(String key, String searchId) {
- if (searchId == null)
- searchId = DEFAULT_SEARCH_ID;
- String effectiveKey = key + "?" + searchId;
- if (!getResultsCache().containsKey(effectiveKey)) {
- return false;
- }
- Cursor cursor = getResultsCache().get(effectiveKey);
- getResultsCache().put(effectiveKey, cursor);
- return true;
- }
-
- private Cursor getResultsSet(String key, Callable<Cursor> makeResultsSet) {
- if (!getResultsCache().containsKey(key)) {
- Cursor resultSet = null;
- try {
- resultSet = makeResultsSet.call();
- } catch (HiveClientException ex) {
- throw new HiveClientFormattedException(ex);
- } catch (Exception ex) {
- throw new ServiceFormattedException(ex.getMessage(), ex);
- }
- getResultsCache().put(key, resultSet);
- }
-
- return getResultsCache().get(key);
- }
-
- public Response.ResponseBuilder request(String key, String searchId, boolean canExpire, String fromBeginning, Integer count, String format, Callable<Cursor> makeResultsSet) throws HiveClientException {
- if (searchId == null)
- searchId = DEFAULT_SEARCH_ID;
- key = key + "?" + searchId;
- if (!canExpire)
- key = "$" + key;
- if (fromBeginning != null && fromBeginning.equals("true") && getResultsCache().containsKey(key))
- getResultsCache().remove(key);
- Cursor resultSet = getResultsSet(key, makeResultsSet);
-
- if (count == null)
- count = DEFAULT_FETCH_COUNT;
-
- ArrayList<ColumnDescription> schema = resultSet.getSchema();
- ArrayList<Object[]> rows = new ArrayList<Object[]>(count);
- int read = resultSet.readRaw(rows, count);
- if(format != null && format.equalsIgnoreCase("d3")) {
- List<Map<String,Object>> results = new ArrayList<Map<String,Object>>();
- for(int i=0; i<rows.size(); i++) {
- Object[] row = rows.get(i);
- Map<String, Object> keyValue = new HashMap<String, Object>(row.length);
- for(int j=0; j<row.length; j++) {
- //Replace dots in schema with underscore
- String schemaName = schema.get(j).getName();
- keyValue.put(schemaName.replace('.','_'), row[j]);
- }
- results.add(keyValue);
- }
- return Response.ok(results);
- } else {
- ResultsResponse resultsResponse = new ResultsResponse();
- resultsResponse.setSchema(schema);
- resultsResponse.setRows(rows);
- resultsResponse.setReadCount(read);
- resultsResponse.setHasNext(resultSet.hasNext());
- // resultsResponse.setSize(resultSet.size());
- resultsResponse.setOffset(resultSet.getOffset());
- resultsResponse.setHasResults(true);
- return Response.ok(resultsResponse);
- }
- }
-
- public static Response.ResponseBuilder emptyResponse() {
- ResultsResponse resultsResponse = new ResultsResponse();
- resultsResponse.setSchema(new ArrayList<ColumnDescription>());
- resultsResponse.setRows(new ArrayList<Object[]>());
- resultsResponse.setReadCount(0);
- resultsResponse.setHasNext(false);
- resultsResponse.setOffset(0);
- resultsResponse.setHasResults(false);
- return Response.ok(resultsResponse);
- }
-
- private static class ResultsResponse {
- private ArrayList<ColumnDescription> schema;
- private ArrayList<String[]> rows;
- private int readCount;
- private boolean hasNext;
- private long offset;
- private boolean hasResults;
-
- public void setSchema(ArrayList<ColumnDescription> schema) {
- this.schema = schema;
- }
-
- public ArrayList<ColumnDescription> getSchema() {
- return schema;
- }
-
- public void setRows(ArrayList<Object[]> rows) {
- if( null == rows ){
- this.rows = null;
- }
- this.rows = new ArrayList<String[]>(rows.size());
- for(Object[] row : rows ){
- String[] strs = new String[row.length];
- for( int colNum = 0 ; colNum < row.length ; colNum++ ){
- String value = String.valueOf(row[colNum]);
- if(row[colNum] != null && (value.isEmpty() || value.equalsIgnoreCase("null"))){
- strs[colNum] = String.format("\"%s\"",value);
- }else{
- strs[colNum] = value;
- }
- }
- this.rows.add(strs);
- }
- }
-
- public ArrayList<String[]> getRows() {
- return rows;
- }
-
- public void setReadCount(int readCount) {
- this.readCount = readCount;
- }
-
- public int getReadCount() {
- return readCount;
- }
-
- public void setHasNext(boolean hasNext) {
- this.hasNext = hasNext;
- }
-
- public boolean isHasNext() {
- return hasNext;
- }
-
- public long getOffset() {
- return offset;
- }
-
- public void setOffset(long offset) {
- this.offset = offset;
- }
-
- public boolean getHasResults() {
- return hasResults;
- }
-
- public void setHasResults(boolean hasResults) {
- this.hasResults = hasResults;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c0f9621f/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/StoredOperationHandle.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/StoredOperationHandle.java b/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/StoredOperationHandle.java
deleted file mode 100644
index a1d87ad..0000000
--- a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/StoredOperationHandle.java
+++ /dev/null
@@ -1,146 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ambari.view.hive.resources.jobs;
-
-import org.apache.ambari.view.hive.persistence.utils.Indexed;
-import org.apache.ambari.view.hive.utils.ServiceFormattedException;
-import org.apache.commons.beanutils.PropertyUtils;
-import org.apache.commons.codec.DecoderException;
-import org.apache.commons.codec.binary.Hex;
-import org.apache.hive.service.cli.thrift.THandleIdentifier;
-import org.apache.hive.service.cli.thrift.TOperationHandle;
-import org.apache.hive.service.cli.thrift.TOperationType;
-
-import java.lang.reflect.InvocationTargetException;
-import java.util.Map;
-
-/**
- * Bean to represent TOperationHandle stored in DB
- */
-public class StoredOperationHandle implements Indexed {
- private boolean hasResultSet;
- private double modifiedRowCount;
- private int operationType;
- private String guid;
- private String secret;
-
- private String jobId;
-
- private String id;
-
- public StoredOperationHandle() {}
- public StoredOperationHandle(Map<String, Object> stringObjectMap) throws InvocationTargetException, IllegalAccessException {
- for (Map.Entry<String, Object> entry : stringObjectMap.entrySet()) {
- try {
- PropertyUtils.setProperty(this, entry.getKey(), entry.getValue());
- } catch (NoSuchMethodException e) {
- //do nothing, skip
- }
- }
- }
-
-
- public static StoredOperationHandle buildFromTOperationHandle(TOperationHandle handle) {
- StoredOperationHandle storedHandle = new StoredOperationHandle();
- //bool hasResultSet
- storedHandle.setHasResultSet(handle.isHasResultSet());
- //optional double modifiedRowCount
- storedHandle.setModifiedRowCount(handle.getModifiedRowCount());
- //TOperationType operationType
- storedHandle.setOperationType(handle.getOperationType().getValue());
- //THandleIdentifier operationId
- storedHandle.setGuid(Hex.encodeHexString(handle.getOperationId().getGuid()));
- storedHandle.setSecret(Hex.encodeHexString(handle.getOperationId().getSecret()));
- return storedHandle;
- }
-
- public TOperationHandle toTOperationHandle() {
- TOperationHandle handle = new TOperationHandle();
- handle.setHasResultSet(isHasResultSet());
- handle.setModifiedRowCount(getModifiedRowCount());
- handle.setOperationType(TOperationType.findByValue(getOperationType()));
- THandleIdentifier identifier = new THandleIdentifier();
- try {
- identifier.setGuid(Hex.decodeHex(getGuid().toCharArray()));
- identifier.setSecret(Hex.decodeHex(getSecret().toCharArray()));
- } catch (DecoderException e) {
- throw new ServiceFormattedException("E060 Wrong identifier of OperationHandle is stored in DB");
- }
- handle.setOperationId(identifier);
- return handle;
- }
-
- public boolean isHasResultSet() {
- return hasResultSet;
- }
-
- public void setHasResultSet(boolean hasResultSet) {
- this.hasResultSet = hasResultSet;
- }
-
- public double getModifiedRowCount() {
- return modifiedRowCount;
- }
-
- public void setModifiedRowCount(double modifiedRowCount) {
- this.modifiedRowCount = modifiedRowCount;
- }
-
- public int getOperationType() {
- return operationType;
- }
-
- public void setOperationType(int operationType) {
- this.operationType = operationType;
- }
-
- public String getGuid() {
- return guid;
- }
-
- public void setGuid(String guid) {
- this.guid = guid;
- }
-
- public String getSecret() {
- return secret;
- }
-
- public void setSecret(String secret) {
- this.secret = secret;
- }
-
- public String getJobId() {
- return jobId;
- }
-
- public void setJobId(String jobId) {
- this.jobId = jobId;
- }
-
- @Override
- public String getId() {
- return id;
- }
-
- @Override
- public void setId(String id) {
- this.id = id;
- }
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c0f9621f/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/ATSParser.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/ATSParser.java b/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/ATSParser.java
deleted file mode 100644
index b145df2..0000000
--- a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/ATSParser.java
+++ /dev/null
@@ -1,248 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ambari.view.hive.resources.jobs.atsJobs;
-
-import org.json.simple.JSONArray;
-import org.json.simple.JSONObject;
-import org.json.simple.JSONValue;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.LinkedList;
-import java.util.List;
-
-/**
- * Parser of ATS responses
- */
-public class ATSParser implements IATSParser {
- protected final static Logger LOG =
- LoggerFactory.getLogger(ATSParser.class);
-
- private ATSRequestsDelegate delegate;
-
- private static final long MillisInSecond = 1000L;
-
- public ATSParser(ATSRequestsDelegate delegate) {
- this.delegate = delegate;
- }
-
- /**
- * returns all HiveQueryIDs from ATS for the given user.
- * @param username
- * @return
- */
- @Override
- public List<HiveQueryId> getHiveQueryIdsForUser(String username) {
- JSONObject entities = delegate.hiveQueryIdsForUser(username);
- return parseHqidJsonFromATS(entities);
- }
-
- /**
- * parses the JSONArray or hive query IDs
- * @param entities: should contain 'entities' element as JSONArray
- * @return
- */
- private List<HiveQueryId> parseHqidJsonFromATS(JSONObject entities) {
- JSONArray jobs = (JSONArray) entities.get("entities");
-
- return getHqidListFromJsonArray(jobs);
- }
-
- /**
- * parses List of HiveQueryIds from JSON
- * @param jobs
- * @return
- */
- private List<HiveQueryId> getHqidListFromJsonArray(JSONArray jobs) {
- List<HiveQueryId> parsedJobs = new LinkedList<>();
- for (Object job : jobs) {
- try {
- HiveQueryId parsedJob = parseAtsHiveJob((JSONObject) job);
- parsedJobs.add(parsedJob);
- } catch (Exception ex) {
- LOG.error("Error while parsing ATS job", ex);
- }
- }
-
- return parsedJobs;
- }
-
- @Override
- public List<TezVertexId> getVerticesForDAGId(String dagId) {
- JSONObject entities = delegate.tezVerticesListForDAG(dagId);
- JSONArray vertices = (JSONArray) entities.get("entities");
-
- List<TezVertexId> parsedVertices = new LinkedList<TezVertexId>();
- for(Object vertex : vertices) {
- try {
- TezVertexId parsedVertex = parseVertex((JSONObject) vertex);
- parsedVertices.add(parsedVertex);
- } catch (Exception ex) {
- LOG.error("Error while parsing the vertex", ex);
- }
- }
-
- return parsedVertices;
- }
-
- @Override
- public HiveQueryId getHiveQueryIdByOperationId(String guidString) {
- JSONObject entities = delegate.hiveQueryIdByOperationId(guidString);
- return getHiveQueryIdFromJson(entities);
- }
-
- private HiveQueryId getHiveQueryIdFromJson(JSONObject entities) {
- JSONArray jobs = (JSONArray) entities.get("entities");
-
- if (jobs.size() == 0) {
- return new HiveQueryId();
- }
-
- return parseAtsHiveJob((JSONObject) jobs.get(0));
- }
-
- /**
- * returns the hive entity from ATS. empty object if not found.
- *
- * @param hiveId: the entityId of the hive
- * @return: empty entity if not found else HiveQueryId
- */
- @Override
- public HiveQueryId getHiveQueryIdByHiveEntityId(String hiveId) {
- JSONObject entity = delegate.hiveQueryEntityByEntityId(hiveId);
- return parseAtsHiveJob(entity);
- }
-
- @Override
- public TezDagId getTezDAGByName(String name) {
- JSONArray tezDagEntities = (JSONArray) delegate.tezDagByName(name).get("entities");
- return parseTezDag(tezDagEntities);
- }
-
- @Override
- public TezDagId getTezDAGByEntity(String entity) {
- JSONArray tezDagEntities = (JSONArray) delegate.tezDagByEntity(entity).get("entities");
- return parseTezDag(tezDagEntities);
- }
-
- /**
- * fetches the HIVE_QUERY_ID from ATS for given user between given time period
- *
- * @param username: username for which to fetch hive query IDs
- * @param startTime: time in miliseconds, inclusive
- * @param endTime: time in miliseconds, exclusive
- * @return: List of HIVE_QUERY_ID
- */
- @Override
- public List<HiveQueryId> getHiveQueryIdsForUserByTime(String username, long startTime, long endTime) {
- JSONObject entities = delegate.hiveQueryIdsForUserByTime(username, startTime, endTime);
- return parseHqidJsonFromATS(entities);
- }
-
- @Override
- public List<HiveQueryId> getHiveQueryIdByEntityList(List<String> hiveIds) {
- List<HiveQueryId> hiveQueryIds = new LinkedList<>();
- for (String id : hiveIds) {
- HiveQueryId hqi = this.getHiveQueryIdByHiveEntityId(id);
- if (null != hqi.entity) {
- hiveQueryIds.add(hqi);
- }
- }
- return hiveQueryIds;
- }
-
- private TezDagId parseTezDag(JSONArray tezDagEntities) {
- assert tezDagEntities.size() <= 1;
- if (tezDagEntities.size() == 0) {
- return new TezDagId();
- }
- JSONObject tezDagEntity = (JSONObject) tezDagEntities.get(0);
-
- TezDagId parsedDag = new TezDagId();
- JSONArray applicationIds = (JSONArray) ((JSONObject) tezDagEntity.get("primaryfilters")).get("applicationId");
- parsedDag.entity = (String) tezDagEntity.get("entity");
- parsedDag.applicationId = (String) applicationIds.get(0);
- parsedDag.status = (String) ((JSONObject) tezDagEntity.get("otherinfo")).get("status");
- return parsedDag;
- }
-
- private HiveQueryId parseAtsHiveJob(JSONObject job) {
- HiveQueryId parsedJob = new HiveQueryId();
-
- parsedJob.entity = (String) job.get("entity");
- parsedJob.url = delegate.hiveQueryIdDirectUrl((String) job.get("entity"));
- parsedJob.starttime = ((Long) job.get("starttime"));
-
- JSONObject primaryfilters = (JSONObject) job.get("primaryfilters");
- JSONArray operationIds = (JSONArray) primaryfilters.get("operationid");
- if (operationIds != null) {
- parsedJob.operationId = (String) (operationIds).get(0);
- }
- JSONArray users = (JSONArray) primaryfilters.get("user");
- if (users != null) {
- parsedJob.user = (String) (users).get(0);
- }
-
- JSONObject lastEvent = getLastEvent(job);
- long lastEventTimestamp = ((Long) lastEvent.get("timestamp"));
-
- parsedJob.duration = (lastEventTimestamp - parsedJob.starttime) / MillisInSecond;
-
- JSONObject otherinfo = (JSONObject) job.get("otherinfo");
- if (otherinfo.get("QUERY") != null) { // workaround for HIVE-10829
- JSONObject query = (JSONObject) JSONValue.parse((String) otherinfo.get("QUERY"));
-
- parsedJob.query = (String) query.get("queryText");
- JSONObject stages = (JSONObject) ((JSONObject) query.get("queryPlan")).get("STAGE PLANS");
-
- List<String> dagIds = new LinkedList<String>();
- List<JSONObject> stagesList = new LinkedList<JSONObject>();
-
- for (Object key : stages.keySet()) {
- JSONObject stage = (JSONObject) stages.get(key);
- if (stage.get("Tez") != null) {
- String dagId = (String) ((JSONObject) stage.get("Tez")).get("DagName:");
- dagIds.add(dagId);
- }
- stagesList.add(stage);
- }
- parsedJob.dagNames = dagIds;
- parsedJob.stages = stagesList;
- }
-
- if (otherinfo.get("VERSION") != null) {
- parsedJob.version = (Long) otherinfo.get("VERSION");
- }
- return parsedJob;
- }
-
- private TezVertexId parseVertex(JSONObject vertex) {
- TezVertexId tezVertexId = new TezVertexId();
- tezVertexId.entity = (String)vertex.get("entity");
- JSONObject otherinfo = (JSONObject)vertex.get("otherinfo");
- if (otherinfo != null)
- tezVertexId.vertexName = (String)otherinfo.get("vertexName");
- return tezVertexId;
- }
-
- private JSONObject getLastEvent(JSONObject atsEntity) {
- JSONArray events = (JSONArray) atsEntity.get("events");
- return (JSONObject) events.get(0);
- }
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c0f9621f/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/ATSParserFactory.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/ATSParserFactory.java b/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/ATSParserFactory.java
deleted file mode 100644
index f035f75..0000000
--- a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/ATSParserFactory.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ambari.view.hive.resources.jobs.atsJobs;
-
-import org.apache.ambari.view.ViewContext;
-import org.apache.ambari.view.utils.ambari.AmbariApi;
-
-import java.util.HashMap;
-import java.util.Map;
-
-public class ATSParserFactory {
-
- private ViewContext context;
- private final AmbariApi ambariApi;
-
- public ATSParserFactory(ViewContext context) {
- this.context = context;
- this.ambariApi = new AmbariApi(context);
- }
-
- public ATSParser getATSParser() {
- ATSRequestsDelegateImpl delegate = new ATSRequestsDelegateImpl(context, getATSUrl());
- return new ATSParser(delegate);
- }
-
- public String getATSUrl() {
- return ambariApi.getServices().getTimelineServerUrl();
- }
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c0f9621f/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/ATSRequestsDelegate.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/ATSRequestsDelegate.java b/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/ATSRequestsDelegate.java
deleted file mode 100644
index 6de5773..0000000
--- a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/ATSRequestsDelegate.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ambari.view.hive.resources.jobs.atsJobs;
-
-import org.json.simple.JSONObject;
-
-public interface ATSRequestsDelegate {
- String hiveQueryIdDirectUrl(String entity);
-
- String hiveQueryIdOperationIdUrl(String operationId);
-
- String tezDagDirectUrl(String entity);
-
- String tezDagNameUrl(String name);
-
- String tezVerticesListForDAGUrl(String dagId);
-
- JSONObject hiveQueryIdsForUser(String username);
-
- JSONObject hiveQueryIdByOperationId(String operationId);
-
- JSONObject tezDagByName(String name);
-
- JSONObject tezVerticesListForDAG(String dagId);
-
- JSONObject tezDagByEntity(String entity);
-
- JSONObject hiveQueryIdsForUserByTime(String username, long startTime, long endTime);
-
- JSONObject hiveQueryEntityByEntityId(String hiveEntityId);
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c0f9621f/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/ATSRequestsDelegateImpl.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/ATSRequestsDelegateImpl.java b/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/ATSRequestsDelegateImpl.java
deleted file mode 100644
index f52c9da..0000000
--- a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/ATSRequestsDelegateImpl.java
+++ /dev/null
@@ -1,175 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ambari.view.hive.resources.jobs.atsJobs;
-
-import org.apache.ambari.view.ViewContext;
-import org.apache.commons.io.IOUtils;
-import org.json.simple.JSONObject;
-import org.json.simple.JSONValue;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.HashMap;
-
-public class ATSRequestsDelegateImpl implements ATSRequestsDelegate {
- protected final static Logger LOG =
- LoggerFactory.getLogger(ATSRequestsDelegateImpl.class);
- public static final String EMPTY_ENTITIES_JSON = "{ \"entities\" : [ ] }";
-
- private ViewContext context;
- private String atsUrl;
-
- public ATSRequestsDelegateImpl(ViewContext context, String atsUrl) {
- this.context = context;
- this.atsUrl = addProtocolIfMissing(atsUrl);
- }
-
- private String addProtocolIfMissing(String atsUrl) {
- if (!atsUrl.matches("^[^:]+://.*$"))
- atsUrl = "http://" + atsUrl;
- return atsUrl;
- }
-
- @Override
- public String hiveQueryIdDirectUrl(String entity) {
- return atsUrl + "/ws/v1/timeline/HIVE_QUERY_ID/" + entity;
- }
-
- @Override
- public String hiveQueryIdOperationIdUrl(String operationId) {
- // ATS parses operationId started with digit as integer and not returns the response.
- // Quotation prevents this.
- return atsUrl + "/ws/v1/timeline/HIVE_QUERY_ID?primaryFilter=operationid:%22" + operationId + "%22";
- }
-
- @Override
- public String tezDagDirectUrl(String entity) {
- return atsUrl + "/ws/v1/timeline/TEZ_DAG_ID/" + entity;
- }
-
- @Override
- public String tezDagNameUrl(String name) {
- return atsUrl + "/ws/v1/timeline/TEZ_DAG_ID?primaryFilter=dagName:" + name;
- }
-
- @Override
- public String tezVerticesListForDAGUrl(String dagId) {
- return atsUrl + "/ws/v1/timeline/TEZ_VERTEX_ID?primaryFilter=TEZ_DAG_ID:" + dagId;
- }
-
- @Override
- public JSONObject hiveQueryIdsForUser(String username) {
- String hiveQueriesListUrl = atsUrl + "/ws/v1/timeline/HIVE_QUERY_ID?primaryFilter=requestuser:" + username;
- String response = readFromWithDefault(hiveQueriesListUrl, "{ \"entities\" : [ ] }");
- return (JSONObject) JSONValue.parse(response);
- }
-
- @Override
- public JSONObject hiveQueryIdByOperationId(String operationId) {
- String hiveQueriesListUrl = hiveQueryIdOperationIdUrl(operationId);
- String response = readFromWithDefault(hiveQueriesListUrl, EMPTY_ENTITIES_JSON);
- return (JSONObject) JSONValue.parse(response);
- }
-
- @Override
- public JSONObject tezDagByName(String name) {
- String tezDagUrl = tezDagNameUrl(name);
- String response = readFromWithDefault(tezDagUrl, EMPTY_ENTITIES_JSON);
- return (JSONObject) JSONValue.parse(response);
- }
-
- @Override
- public JSONObject tezDagByEntity(String entity) {
- String tezDagEntityUrl = tezDagEntityUrl(entity);
- String response = readFromWithDefault(tezDagEntityUrl, EMPTY_ENTITIES_JSON);
- return (JSONObject) JSONValue.parse(response);
- }
-
- /**
- * fetches the HIVE_QUERY_ID from ATS for given user between given time period
- * @param username: username for which to fetch hive query IDs
- * @param startTime: time in miliseconds, inclusive
- * @param endTime: time in miliseconds, exclusive
- * @return
- */
- @Override
- public JSONObject hiveQueryIdsForUserByTime(String username, long startTime, long endTime) {
- StringBuilder url = new StringBuilder();
- url.append(atsUrl).append("/ws/v1/timeline/HIVE_QUERY_ID?")
- .append("windowStart=").append(startTime)
- .append("&windowEnd=").append(endTime)
- .append("&primaryFilter=requestuser:").append(username);
- String hiveQueriesListUrl = url.toString();
-
- String response = readFromWithDefault(hiveQueriesListUrl, EMPTY_ENTITIES_JSON);
- return (JSONObject) JSONValue.parse(response);
- }
-
- @Override
- public JSONObject hiveQueryEntityByEntityId(String hiveEntityId) {
- StringBuilder url = new StringBuilder();
- url.append(atsUrl).append("/ws/v1/timeline/HIVE_QUERY_ID/").append(hiveEntityId);
- String hiveQueriesListUrl = url.toString();
- String response = readFromWithDefault(hiveQueriesListUrl, EMPTY_ENTITIES_JSON);
- return (JSONObject) JSONValue.parse(response);
- }
-
- private String tezDagEntityUrl(String entity) {
- return atsUrl + "/ws/v1/timeline/TEZ_DAG_ID?primaryFilter=callerId:" + entity;
- }
-
- public boolean checkATSStatus() throws IOException {
- String url = atsUrl + "/ws/v1/timeline/";
- InputStream responseInputStream = context.getURLStreamProvider().readAsCurrent(url, "GET",
- (String)null, new HashMap<String, String>());
- IOUtils.toString(responseInputStream);
- return true;
- }
-
- @Override
- public JSONObject tezVerticesListForDAG(String dagId) {
- String response = readFromWithDefault(tezVerticesListForDAGUrl(dagId), "{ \"entities\" : [ ] }");
- return (JSONObject) JSONValue.parse(response);
- }
-
-
-
- protected String readFromWithDefault(String atsUrl, String defaultResponse) {
- String response;
- try {
- InputStream responseInputStream = context.getURLStreamProvider().readAsCurrent(atsUrl, "GET",
- (String)null, new HashMap<String, String>());
- response = IOUtils.toString(responseInputStream);
- } catch (IOException e) {
- LOG.error("Error while reading from ATS", e);
- response = defaultResponse;
- }
- return response;
- }
-
- public String getAtsUrl() {
- return atsUrl;
- }
-
- public void setAtsUrl(String atsUrl) {
- this.atsUrl = atsUrl;
- }
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c0f9621f/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/HiveQueryId.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/HiveQueryId.java b/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/HiveQueryId.java
deleted file mode 100644
index bb81fef..0000000
--- a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/HiveQueryId.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ambari.view.hive.resources.jobs.atsJobs;
-
-import org.json.simple.JSONObject;
-
-import java.util.List;
-
-public class HiveQueryId {
- public static long ATS_15_RESPONSE_VERSION = 2; // version returned from ATS 1.5 release
-
- public String url;
-
- public String entity;
- public String query;
-
- public List<String> dagNames;
-
- public List<JSONObject> stages;
-
- public long starttime;
- public long duration;
- public String operationId;
- public String user;
- public long version;
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c0f9621f/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/IATSParser.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/IATSParser.java b/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/IATSParser.java
deleted file mode 100644
index 547dfec..0000000
--- a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/IATSParser.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ambari.view.hive.resources.jobs.atsJobs;
-
-import java.util.List;
-
-public interface IATSParser {
- List<HiveQueryId> getHiveQueryIdsForUser(String username);
-
- List<TezVertexId> getVerticesForDAGId(String dagId);
-
- HiveQueryId getHiveQueryIdByOperationId(String guidString);
-
- TezDagId getTezDAGByName(String name);
-
- TezDagId getTezDAGByEntity(String entity);
-
- List<HiveQueryId> getHiveQueryIdsForUserByTime(String username, long startTime, long endTime);
-
- HiveQueryId getHiveQueryIdByHiveEntityId(String hiveEntityId);
-
- List<HiveQueryId> getHiveQueryIdByEntityList(List<String> hiveEntityIds);
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c0f9621f/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/TezDagId.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/TezDagId.java b/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/TezDagId.java
deleted file mode 100644
index a814286..0000000
--- a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/TezDagId.java
+++ /dev/null
@@ -1,26 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ambari.view.hive.resources.jobs.atsJobs;
-
-public class TezDagId {
- public static final String STATUS_UNKNOWN = "UNKNOWN";
- public String applicationId = "";
- public String entity = "";
- public String status = STATUS_UNKNOWN;
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c0f9621f/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/TezVertexId.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/TezVertexId.java b/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/TezVertexId.java
deleted file mode 100644
index 34dd996..0000000
--- a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/atsJobs/TezVertexId.java
+++ /dev/null
@@ -1,24 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ambari.view.hive.resources.jobs.atsJobs;
-
-public class TezVertexId {
- public String entity;
- public String vertexName;
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c0f9621f/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/rm/RMParser.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/rm/RMParser.java b/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/rm/RMParser.java
deleted file mode 100644
index b39be44..0000000
--- a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/rm/RMParser.java
+++ /dev/null
@@ -1,129 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ambari.view.hive.resources.jobs.rm;
-
-import org.apache.ambari.view.hive.resources.jobs.atsJobs.TezVertexId;
-import org.json.simple.JSONArray;
-import org.json.simple.JSONObject;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-
-/**
- * Parser of Resource Manager responses
- */
-public class RMParser {
- protected final static Logger LOG =
- LoggerFactory.getLogger(RMParser.class);
- private RMRequestsDelegate delegate;
-
- public RMParser(RMRequestsDelegate delegate) {
- this.delegate = delegate;
- }
-
- /**
- * Progress of DAG
- * @param appId App Id
- * @param dagId DAG Id
- * @return progress of DAG
- */
- public Double getDAGProgress(String appId, String dagId) {
- String dagIdx = parseDagIdIndex(dagId);
- JSONObject progresses = delegate.dagProgress(appId, dagIdx);
-
- double dagProgressValue;
- if (progresses != null) {
- JSONObject dagProgress = (JSONObject) progresses.get("dagProgress");
- dagProgressValue = (Double) (dagProgress.get("progress"));
- } else {
- LOG.error("Error while retrieving progress of " + appId + ":" + dagId + ". 0 assumed.");
- dagProgressValue = 0;
- }
- return dagProgressValue;
- }
-
- /**
- * Progress of vertices
- * @param appId App Id
- * @param dagId DAG Id
- * @param vertices vertices list
- * @return list of vertices
- */
- public List<VertexProgress> getDAGVerticesProgress(String appId, String dagId, List<TezVertexId> vertices) {
- String dagIdx = parseDagIdIndex(dagId);
-
- Map<String, String> vertexIdToEntityMapping = new HashMap<String, String>();
- StringBuilder builder = new StringBuilder();
- if (vertices.size() > 0) {
- for (TezVertexId vertexId : vertices) {
- String[] parts = vertexId.entity.split("_");
- String vertexIdx = parts[parts.length - 1];
- builder.append(vertexIdx).append(",");
-
- vertexIdToEntityMapping.put(vertexId.entity, vertexId.vertexName);
- }
- builder.setLength(builder.length() - 1); // remove last comma
- }
-
- String commaSeparatedVertices = builder.toString();
-
- List<VertexProgress> parsedVertexProgresses = new LinkedList<VertexProgress>();
- JSONObject vertexProgressesResponse = delegate.verticesProgress(
- appId, dagIdx, commaSeparatedVertices);
- if (vertexProgressesResponse == null) {
- LOG.error("Error while retrieving progress of vertices " +
- appId + ":" + dagId + ":" + commaSeparatedVertices + ". 0 assumed for all vertices.");
- for (TezVertexId vertexId : vertices) {
- VertexProgress vertexProgressInfo = new VertexProgress();
- vertexProgressInfo.name = vertexId.vertexName;
- vertexProgressInfo.progress = 0.0;
- parsedVertexProgresses.add(vertexProgressInfo);
- }
- return parsedVertexProgresses;
- }
- JSONArray vertexProgresses = (JSONArray) vertexProgressesResponse.get("vertexProgresses");
-
- for (Object vertex : vertexProgresses) {
- JSONObject jsonObject = (JSONObject) vertex;
-
- VertexProgress vertexProgressInfo = new VertexProgress();
- vertexProgressInfo.id = (String) jsonObject.get("id");
- vertexProgressInfo.name = vertexIdToEntityMapping.get(vertexProgressInfo.id);
- vertexProgressInfo.progress = (Double) jsonObject.get("progress");
-
- parsedVertexProgresses.add(vertexProgressInfo);
- }
- return parsedVertexProgresses;
- }
-
- public String parseDagIdIndex(String dagId) {
- String[] dagIdParts = dagId.split("_");
- return dagIdParts[dagIdParts.length - 1];
- }
-
- public static class VertexProgress {
- public String id;
- public String name;
- public Double progress;
- }
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c0f9621f/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/rm/RMParserFactory.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/rm/RMParserFactory.java b/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/rm/RMParserFactory.java
deleted file mode 100644
index 260b464..0000000
--- a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/rm/RMParserFactory.java
+++ /dev/null
@@ -1,49 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ambari.view.hive.resources.jobs.rm;
-
-import org.apache.ambari.view.ViewContext;
-import org.apache.ambari.view.hive.utils.ServiceFormattedException;
-import org.apache.ambari.view.utils.ambari.AmbariApi;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class RMParserFactory {
- protected final static Logger LOG =
- LoggerFactory.getLogger(RMParserFactory.class);
-
- private final ViewContext context;
- private final AmbariApi ambariApi;
-
- public RMParserFactory(ViewContext context) {
- this.context = context;
- this.ambariApi = new AmbariApi(context);
- }
-
- public RMParser getRMParser() {
- String rmUrl = getRMUrl();
-
- RMRequestsDelegate delegate = new RMRequestsDelegateImpl(context, rmUrl);
- return new RMParser(delegate);
- }
-
- public String getRMUrl() {
- return ambariApi.getServices().getRMUrl();
- }
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c0f9621f/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/rm/RMRequestsDelegate.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/rm/RMRequestsDelegate.java b/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/rm/RMRequestsDelegate.java
deleted file mode 100644
index 6114b34..0000000
--- a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/rm/RMRequestsDelegate.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ambari.view.hive.resources.jobs.rm;
-
-import org.json.simple.JSONObject;
-
-public interface RMRequestsDelegate {
- String dagProgressUrl(String appId, String dagIdx);
-
- String verticesProgressUrl(String appId, String dagIdx, String vertices);
-
- JSONObject dagProgress(String appId, String dagIdx);
-
- JSONObject verticesProgress(String appId, String dagIdx, String commaSeparatedVertices);
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c0f9621f/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/rm/RMRequestsDelegateImpl.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/rm/RMRequestsDelegateImpl.java b/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/rm/RMRequestsDelegateImpl.java
deleted file mode 100644
index 087ef68..0000000
--- a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/rm/RMRequestsDelegateImpl.java
+++ /dev/null
@@ -1,100 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ambari.view.hive.resources.jobs.rm;
-
-import org.apache.ambari.view.ViewContext;
-import org.apache.ambari.view.hive.utils.ServiceFormattedException;
-import org.apache.ambari.view.utils.ambari.AmbariApi;
-import org.apache.commons.io.IOUtils;
-import org.json.simple.JSONObject;
-import org.json.simple.JSONValue;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.HashMap;
-
-public class RMRequestsDelegateImpl implements RMRequestsDelegate {
- protected final static Logger LOG =
- LoggerFactory.getLogger(RMRequestsDelegateImpl.class);
- public static final String EMPTY_ENTITIES_JSON = "{ \"entities\" : [ ] }";
-
- private ViewContext context;
- private String rmUrl;
-
- public RMRequestsDelegateImpl(ViewContext context, String rmUrl) {
- this.context = context;
- this.rmUrl = rmUrl;
- }
-
- @Override
- public String dagProgressUrl(String appId, String dagIdx) {
- return rmUrl + String.format("/proxy/%s/ws/v1/tez/dagProgress?dagID=%s", appId, dagIdx);
- }
-
- @Override
- public String verticesProgressUrl(String appId, String dagIdx, String vertices) {
- return rmUrl + String.format("/proxy/%s/ws/v1/tez/vertexProgresses?dagID=%s&vertexID=%s", appId, dagIdx, vertices);
- }
-
- @Override
- public JSONObject dagProgress(String appId, String dagIdx) {
- String url = dagProgressUrl(appId, dagIdx);
- String response;
- try {
- InputStream responseInputStream = context.getURLStreamProvider().readFrom(url, "GET",
- (String)null, new HashMap<String, String>());
- response = IOUtils.toString(responseInputStream);
- } catch (IOException e) {
- throw new ServiceFormattedException(
- String.format("R010 DAG %s in app %s not found or ResourceManager is unreachable", dagIdx, appId));
- }
- return (JSONObject) JSONValue.parse(response);
- }
-
- @Override
- public JSONObject verticesProgress(String appId, String dagIdx, String commaSeparatedVertices) {
- String url = verticesProgressUrl(appId, dagIdx, commaSeparatedVertices);
- String response;
- try {
- InputStream responseInputStream = context.getURLStreamProvider().readFrom(url, "GET",
- (String)null, new HashMap<String, String>());
- response = IOUtils.toString(responseInputStream);
- } catch (IOException e) {
- throw new ServiceFormattedException(
- String.format("R020 DAG %s in app %s not found or ResourceManager is unreachable", dagIdx, appId));
- }
- return (JSONObject) JSONValue.parse(response);
- }
-
- protected String readFromWithDefault(String url, String defaultResponse) {
- String response;
- try {
- InputStream responseInputStream = context.getURLStreamProvider().readFrom(url, "GET",
- (String)null, new HashMap<String, String>());
- response = IOUtils.toString(responseInputStream);
- } catch (IOException e) {
- LOG.error("Error while reading from RM", e);
- response = defaultResponse;
- }
- return response;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c0f9621f/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/viewJobs/IJobControllerFactory.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/viewJobs/IJobControllerFactory.java b/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/viewJobs/IJobControllerFactory.java
deleted file mode 100644
index 89fbb85..0000000
--- a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/viewJobs/IJobControllerFactory.java
+++ /dev/null
@@ -1,23 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ambari.view.hive.resources.jobs.viewJobs;
-
-public interface IJobControllerFactory {
- JobController createControllerForJob(Job job);
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c0f9621f/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/viewJobs/Job.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/viewJobs/Job.java b/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/viewJobs/Job.java
deleted file mode 100644
index 98d6589..0000000
--- a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/viewJobs/Job.java
+++ /dev/null
@@ -1,130 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ambari.view.hive.resources.jobs.viewJobs;
-
-
-import org.apache.ambari.view.hive.persistence.utils.Indexed;
-import org.apache.ambari.view.hive.persistence.utils.PersonalResource;
-
-import java.beans.Transient;
-import java.io.Serializable;
-
-/**
- * Interface for Job bean to create Proxy for it
- */
-public interface Job extends Serializable,Indexed,PersonalResource {
- String JOB_STATE_UNKNOWN = "Unknown";
- String JOB_STATE_INITIALIZED = "Initialized";
- String JOB_STATE_RUNNING = "Running";
- String JOB_STATE_FINISHED = "Succeeded";
- String JOB_STATE_CANCELED = "Canceled";
- String JOB_STATE_CLOSED = "Closed";
- String JOB_STATE_ERROR = "Error";
- String JOB_STATE_PENDING = "Pending";
-
- @Transient
- String getHiveQueryId();
-
- @Transient
- void setHiveQueryId(String hiveQueryId);
-
- String getId();
-
- void setId(String id);
-
- String getOwner();
-
- void setOwner(String owner);
-
- String getTitle();
-
- void setTitle(String title);
-
- String getQueryFile();
-
- void setQueryFile(String queryFile);
-
- Long getDateSubmitted();
-
- void setDateSubmitted(Long dateSubmitted);
-
- Long getDuration();
-
- void setDuration(Long duration);
-
- String getStatus();
-
- void setStatus(String status);
-
- String getForcedContent();
-
- void setForcedContent(String forcedContent);
-
- String getQueryId();
-
- void setQueryId(String queryId);
-
- String getStatusDir();
-
- void setStatusDir(String statusDir);
-
- String getDataBase();
-
- void setDataBase(String dataBase);
-
- String getLogFile();
-
- void setLogFile(String logFile);
-
- String getConfFile();
-
- void setConfFile(String confFile);
-
- String getApplicationId();
-
- void setApplicationId(String applicationId);
-
- String getDagName();
-
- void setDagName(String dagName);
-
- String getDagId();
-
- void setDagId(String dagId);
-
- String getSessionTag();
-
- void setSessionTag(String sessionTag);
-
- String getSqlState();
-
- void setSqlState(String sqlState);
-
- String getStatusMessage();
-
- void setStatusMessage(String message);
-
- String getReferrer();
-
- void setReferrer(String referrer);
-
- String getGlobalSettings();
-
- void setGlobalSettings(String globalSettings);
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c0f9621f/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/viewJobs/JobController.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/viewJobs/JobController.java b/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/viewJobs/JobController.java
deleted file mode 100644
index 3922839..0000000
--- a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/viewJobs/JobController.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ambari.view.hive.resources.jobs.viewJobs;
-
-import org.apache.ambari.view.hive.client.Cursor;
-import org.apache.ambari.view.hive.client.HiveClientException;
-import org.apache.ambari.view.hive.persistence.utils.ItemNotFound;
-import org.apache.ambari.view.hive.resources.jobs.NoOperationStatusSetException;
-import org.apache.ambari.view.hive.resources.jobs.OperationHandleController;
-
-public interface JobController {
-
- OperationHandleController.OperationStatus getStatus() throws ItemNotFound, HiveClientException, NoOperationStatusSetException;
-
- void submit();
-
- void cancel() throws ItemNotFound;
-
- Job getJob();
-
- /**
- * Use carefully. Returns unproxied bean object
- * @return unproxied bean object
- */
- Job getJobPOJO();
-
- Cursor getResults() throws ItemNotFound;
- boolean hasResults() throws ItemNotFound;
-
- void afterCreation();
-
- void update();
-
- boolean isModified();
-
- void clearModified();
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c0f9621f/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/viewJobs/JobControllerFactory.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/viewJobs/JobControllerFactory.java b/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/viewJobs/JobControllerFactory.java
deleted file mode 100644
index a2790d1..0000000
--- a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/viewJobs/JobControllerFactory.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ambari.view.hive.resources.jobs.viewJobs;
-
-import org.apache.ambari.view.ViewContext;
-import org.apache.ambari.view.hive.utils.SharedObjectsFactory;
-
-public class JobControllerFactory implements IJobControllerFactory {
- private SharedObjectsFactory sharedObjectsFactory;
- private ViewContext context;
-
- public JobControllerFactory(ViewContext context, SharedObjectsFactory sharedObjectsFactory) {
- this.sharedObjectsFactory = sharedObjectsFactory;
- this.context = context;
- }
-
- @Override
- public JobController createControllerForJob(Job job) {
- return new JobControllerImpl(context, job,
- sharedObjectsFactory.getOperationHandleControllerFactory(),
- sharedObjectsFactory.getSavedQueryResourceManager(),
- sharedObjectsFactory.getATSParser(),
- sharedObjectsFactory.getHdfsApi());
- }
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/c0f9621f/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/viewJobs/JobControllerImpl.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/viewJobs/JobControllerImpl.java b/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/viewJobs/JobControllerImpl.java
deleted file mode 100644
index a408619..0000000
--- a/contrib/views/hive/src/main/java/org/apache/ambari/view/hive/resources/jobs/viewJobs/JobControllerImpl.java
+++ /dev/null
@@ -1,401 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ambari.view.hive.resources.jobs.viewJobs;
-
-import org.apache.ambari.view.ViewContext;
-import org.apache.ambari.view.hive.client.Cursor;
-import org.apache.ambari.view.hive.client.HiveClientException;
-import org.apache.ambari.view.hive.client.HiveClientRuntimeException;
-import org.apache.ambari.view.hive.client.HiveErrorStatusException;
-import org.apache.ambari.view.hive.client.UserLocalConnection;
-import org.apache.ambari.view.hive.persistence.utils.ItemNotFound;
-import org.apache.ambari.view.hive.resources.jobs.ConnectionController;
-import org.apache.ambari.view.hive.resources.jobs.LogParser;
-import org.apache.ambari.view.hive.resources.jobs.ModifyNotificationDelegate;
-import org.apache.ambari.view.hive.resources.jobs.ModifyNotificationInvocationHandler;
-import org.apache.ambari.view.hive.resources.jobs.NoOperationStatusSetException;
-import org.apache.ambari.view.hive.resources.jobs.OperationHandleController;
-import org.apache.ambari.view.hive.resources.jobs.OperationHandleControllerFactory;
-import org.apache.ambari.view.hive.resources.jobs.atsJobs.IATSParser;
-import org.apache.ambari.view.hive.resources.savedQueries.SavedQuery;
-import org.apache.ambari.view.hive.resources.savedQueries.SavedQueryResourceManager;
-import org.apache.ambari.view.hive.utils.BadRequestFormattedException;
-import org.apache.ambari.view.hive.utils.FilePaginator;
-import org.apache.ambari.view.hive.utils.HiveClientFormattedException;
-import org.apache.ambari.view.hive.utils.MisconfigurationFormattedException;
-import org.apache.ambari.view.hive.utils.ServiceFormattedException;
-import org.apache.ambari.view.utils.hdfs.HdfsApi;
-import org.apache.ambari.view.utils.hdfs.HdfsApiException;
-import org.apache.ambari.view.utils.hdfs.HdfsUtil;
-import org.apache.hive.service.cli.thrift.TSessionHandle;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.lang.reflect.Proxy;
-import java.text.SimpleDateFormat;
-import java.util.Date;
-
-public class JobControllerImpl implements JobController, ModifyNotificationDelegate {
- private final static Logger LOG =
- LoggerFactory.getLogger(JobControllerImpl.class);
-
- private ViewContext context;
- private HdfsApi hdfsApi;
- private Job jobUnproxied;
- private Job job;
- private boolean modified;
-
- private OperationHandleControllerFactory opHandleControllerFactory;
- private ConnectionController hiveConnection;
- private SavedQueryResourceManager savedQueryResourceManager;
- private IATSParser atsParser;
-
- /**
- * JobController constructor
- * Warning: Create JobControllers ONLY using JobControllerFactory!
- */
- public JobControllerImpl(ViewContext context, Job job,
- OperationHandleControllerFactory opHandleControllerFactory,
- SavedQueryResourceManager savedQueryResourceManager,
- IATSParser atsParser,
- HdfsApi hdfsApi) {
- this.context = context;
- setJobPOJO(job);
- this.opHandleControllerFactory = opHandleControllerFactory;
- this.savedQueryResourceManager = savedQueryResourceManager;
- this.atsParser = atsParser;
- this.hdfsApi = hdfsApi;
-
- UserLocalConnection connectionLocal = new UserLocalConnection();
- this.hiveConnection = new ConnectionController(opHandleControllerFactory, connectionLocal.get(context));
- }
-
- public String getQueryForJob() {
- FilePaginator paginator = new FilePaginator(job.getQueryFile(), hdfsApi);
- String query;
- try {
- query = paginator.readPage(0); //warning - reading only 0 page restricts size of query to 1MB
- } catch (IOException e) {
- throw new ServiceFormattedException("F030 Error when reading file " + job.getQueryFile(), e);
- } catch (InterruptedException e) {
- throw new ServiceFormattedException("F030 Error when reading file " + job.getQueryFile(), e);
- }
- return query;
- }
-
- private static final String DEFAULT_DB = "default";
- public String getJobDatabase() {
- if (job.getDataBase() != null) {
- return job.getDataBase();
- } else {
- return DEFAULT_DB;
- }
- }
-
- @Override
- public OperationHandleController.OperationStatus getStatus() throws ItemNotFound, HiveClientException, NoOperationStatusSetException {
- OperationHandleController handle = opHandleControllerFactory.getHandleForJob(job);
- return handle.getOperationStatus();
- }
-
- @Override
- public void submit() {
- setupHiveBeforeQueryExecute();
-
- String query = getQueryForJob();
- OperationHandleController handleController = hiveConnection.executeQuery(getSession(), query);
-
- handleController.persistHandleForJob(job);
- }
-
- private void setupHiveBeforeQueryExecute() {
- String database = getJobDatabase();
- hiveConnection.selectDatabase(getSession(), database);
- }
-
- private TSessionHandle getSession() {
- try {
- if (job.getSessionTag() != null) {
- return hiveConnection.getSessionByTag(getJob().getSessionTag());
- }
- } catch (HiveClientException ignore) {
- LOG.debug("Stale sessionTag was provided, new session will be opened");
- }
-
- String tag = hiveConnection.openSession();
- job.setSessionTag(tag);
- try {
- return hiveConnection.getSessionByTag(tag);
- } catch (HiveClientException e) {
- throw new HiveClientFormattedException(e);
- }
- }
-
- @Override
- public void cancel() throws ItemNotFound {
- OperationHandleController handle = opHandleControllerFactory.getHandleForJob(job);
- handle.cancel();
- }
-
- @Override
- public void update() {
- updateOperationStatus();
- updateOperationLogs();
-
- updateJobDuration();
- }
-
- public void updateOperationStatus() {
- try {
-
- OperationHandleController handle = opHandleControllerFactory.getHandleForJob(job);
- OperationHandleController.OperationStatus status = handle.getOperationStatus();
- job.setStatus(status.status);
- job.setStatusMessage(status.message);
- job.setSqlState(status.sqlState);
- LOG.debug("Status of job#" + job.getId() + " is " + job.getStatus());
-
- } catch (NoOperationStatusSetException e) {
- LOG.info("Operation state is not set for job#" + job.getId());
-
- } catch (HiveErrorStatusException e) {
- LOG.debug("Error updating status for job#" + job.getId() + ": " + e.getMessage());
- job.setStatus(Job.JOB_STATE_UNKNOWN);
-
- } catch (HiveClientException e) {
- throw new HiveClientFormattedException(e);
-
- } catch (ItemNotFound itemNotFound) {
- LOG.debug("No TOperationHandle for job#" + job.getId() + ", can't update status");
- }
- }
-
- public void updateOperationLogs() {
- try {
- OperationHandleController handle = opHandleControllerFactory.getHandleForJob(job);
- String logs = handle.getLogs();
-
- LogParser info = LogParser.parseLog(logs);
- LogParser.AppId app = info.getLastAppInList();
- if (app != null) {
- job.setApplicationId(app.getIdentifier());
- }
-
- String logFilePath = job.getLogFile();
- HdfsUtil.putStringToFile(hdfsApi, logFilePath, logs);
-
- } catch (HiveClientRuntimeException ex) {
- LOG.error("Error while fetching logs: " + ex.getMessage());
- } catch (ItemNotFound itemNotFound) {
- LOG.debug("No TOperationHandle for job#" + job.getId() + ", can't read logs");
- } catch (HdfsApiException e) {
- throw new ServiceFormattedException(e);
- }
- }
-
- public boolean isJobEnded() {
- String status = job.getStatus();
- return status.equals(Job.JOB_STATE_FINISHED) || status.equals(Job.JOB_STATE_CANCELED) ||
- status.equals(Job.JOB_STATE_CLOSED) || status.equals(Job.JOB_STATE_ERROR) ||
- status.equals(Job.JOB_STATE_UNKNOWN); // Unknown is not finished, but polling makes no sense
- }
-
- @Override
- public Job getJob() {
- return job;
- }
-
- /**
- * Use carefully. Returns unproxied bean object
- * @return unproxied bean object
- */
- @Override
- public Job getJobPOJO() {
- return jobUnproxied;
- }
-
- public void setJobPOJO(Job jobPOJO) {
- Job jobModifyNotificationProxy = (Job) Proxy.newProxyInstance(jobPOJO.getClass().getClassLoader(),
- new Class[]{Job.class},
- new ModifyNotificationInvocationHandler(jobPOJO, this));
- this.job = jobModifyNotificationProxy;
-
- this.jobUnproxied = jobPOJO;
- }
-
- @Override
- public Cursor getResults() throws ItemNotFound {
- OperationHandleController handle = opHandleControllerFactory.getHandleForJob(job);
- return handle.getResults();
- }
-
- @Override
- public boolean hasResults() throws ItemNotFound {
- OperationHandleController handle = opHandleControllerFactory.getHandleForJob(job);
- return handle.hasResults();
- }
-
- @Override
- public void afterCreation() {
- setupStatusDirIfNotPresent();
- setupQueryFileIfNotPresent();
- setupLogFileIfNotPresent();
-
- setCreationDate();
- }
-
- public void setupLogFileIfNotPresent() {
- if (job.getLogFile() == null || job.getLogFile().isEmpty()) {
- setupLogFile();
- }
- }
-
- public void setupQueryFileIfNotPresent() {
- if (job.getQueryFile() == null || job.getQueryFile().isEmpty()) {
- setupQueryFile();
- }
- }
-
- public void setupStatusDirIfNotPresent() {
- if (job.getStatusDir() == null || job.getStatusDir().isEmpty()) {
- setupStatusDir();
- }
- }
-
- private static final long MillisInSecond = 1000L;
-
- public void updateJobDuration() {
- job.setDuration((System.currentTimeMillis() / MillisInSecond) - (job.getDateSubmitted() / MillisInSecond));
- }
-
- public void setCreationDate() {
- job.setDateSubmitted(System.currentTimeMillis());
- }
-
- private void setupLogFile() {
- LOG.debug("Creating log file for job#" + job.getId());
-
- String logFile = job.getStatusDir() + "/" + "logs";
- try {
- HdfsUtil.putStringToFile(hdfsApi, logFile, "");
- } catch (HdfsApiException e) {
- throw new ServiceFormattedException(e);
- }
-
- job.setLogFile(logFile);
- LOG.debug("Log file for job#" + job.getId() + ": " + logFile);
- }
-
- private void setupStatusDir() {
- String newDirPrefix = makeStatusDirectoryPrefix();
- String newDir = null;
- try {
- newDir = HdfsUtil.findUnallocatedFileName(hdfsApi, newDirPrefix, "");
- } catch (HdfsApiException e) {
- throw new ServiceFormattedException(e);
- }
-
- job.setStatusDir(newDir);
- LOG.debug("Status dir for job#" + job.getId() + ": " + newDir);
- }
-
- private String makeStatusDirectoryPrefix() {
- String userScriptsPath = context.getProperties().get("jobs.dir");
-
- if (userScriptsPath == null) { // TODO: move check to initialization code
- String msg = "jobs.dir is not configured!";
- LOG.error(msg);
- throw new MisconfigurationFormattedException("jobs.dir");
- }
-
- String normalizedName = String.format("hive-job-%s", job.getId());
- String timestamp = new SimpleDateFormat("yyyy-MM-dd_hh-mm").format(new Date());
- return String.format(userScriptsPath +
- "/%s-%s", normalizedName, timestamp);
- }
-
- private void setupQueryFile() {
- String statusDir = job.getStatusDir();
- assert statusDir != null : "setupStatusDir() should be called first";
-
- String jobQueryFilePath = statusDir + "/" + "query.hql";
-
- try {
-
- if (job.getForcedContent() != null) {
-
- HdfsUtil.putStringToFile(hdfsApi, jobQueryFilePath, job.getForcedContent());
- job.setForcedContent(""); // prevent forcedContent to be written to DB
-
- }
- else if (job.getQueryId() != null) {
-
- String savedQueryFile = getRelatedSavedQueryFile();
- hdfsApi.copy(savedQueryFile, jobQueryFilePath);
- job.setQueryFile(jobQueryFilePath);
-
- } else {
-
- throw new BadRequestFormattedException("queryId or forcedContent should be passed!", null);
-
- }
-
- } catch (IOException e) {
- throw new ServiceFormattedException("F040 Error when creating file " + jobQueryFilePath, e);
- } catch (InterruptedException e) {
- throw new ServiceFormattedException("F040 Error when creating file " + jobQueryFilePath, e);
- } catch (HdfsApiException e) {
- throw new ServiceFormattedException(e);
- }
- job.setQueryFile(jobQueryFilePath);
-
- LOG.debug("Query file for job#" + job.getId() + ": " + jobQueryFilePath);
- }
-
- private String getRelatedSavedQueryFile() {
- SavedQuery savedQuery;
- try {
- savedQuery = savedQueryResourceManager.read(job.getQueryId());
- } catch (ItemNotFound itemNotFound) {
- throw new BadRequestFormattedException("queryId not found!", itemNotFound);
- }
- return savedQuery.getQueryFile();
- }
-
- @Override
- public boolean onModification(Object object) {
- setModified(true);
- return true;
- }
-
- @Override
- public boolean isModified() {
- return modified;
- }
-
- public void setModified(boolean modified) {
- this.modified = modified;
- }
-
- @Override
- public void clearModified() {
- setModified(false);
- }
-}