You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ambari.apache.org by jl...@apache.org on 2016/06/27 23:36:42 UTC
[16/34] ambari git commit: AMBARI-17355 & AMBARI-17354: POC: FE & BE
changes for first class support for Yarn hosted services
http://git-wip-us.apache.org/repos/asf/ambari/blob/b88db3cc/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/PersonalCRUDResourceManager.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/PersonalCRUDResourceManager.java b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/PersonalCRUDResourceManager.java
new file mode 100644
index 0000000..2c774df
--- /dev/null
+++ b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/PersonalCRUDResourceManager.java
@@ -0,0 +1,99 @@
+/**
+ * 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.hive2.resources;
+
+import org.apache.ambari.view.ViewContext;
+import org.apache.ambari.view.hive2.persistence.IStorageFactory;
+import org.apache.ambari.view.hive2.persistence.utils.ItemNotFound;
+import org.apache.ambari.view.hive2.persistence.utils.PersonalResource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Callable;
+
+/**
+ * Resource manager that returns only user owned elements from DB
+ * @param <T> Data type with ID and Owner
+ */
+public class PersonalCRUDResourceManager<T extends PersonalResource> extends CRUDResourceManager<T> {
+ protected boolean ignorePermissions = false;
+
+ private final static Logger LOG =
+ LoggerFactory.getLogger(PersonalCRUDResourceManager.class);
+ protected ViewContext context;
+
+ /**
+ * Constructor
+ * @param resourceClass model class
+ */
+ public PersonalCRUDResourceManager(Class<? extends T> resourceClass, IStorageFactory storageFabric, ViewContext context) {
+ super(resourceClass, storageFabric);
+ this.context = context;
+ }
+
+ @Override
+ public T update(T newObject, String id) throws ItemNotFound {
+ T object = storageFactory.getStorage().load(this.resourceClass, id);
+ if (object.getOwner().compareTo(this.context.getUsername()) != 0) {
+ throw new ItemNotFound();
+ }
+
+ newObject.setOwner(this.context.getUsername());
+ return super.update(newObject, id);
+ }
+
+ @Override
+ public T save(T object) {
+ if (!ignorePermissions) {
+ // in threads permissions should be ignored,
+ // because context.getUsername doesn't work. See BUG-27093.
+ object.setOwner(this.context.getUsername());
+ }
+ return super.save(object);
+ }
+
+ @Override
+ protected boolean checkPermissions(T object) {
+ if (ignorePermissions) {
+ return true;
+ }
+ return object.getOwner().compareTo(this.context.getUsername()) == 0;
+ }
+
+ /**
+ * Execute action ignoring objects owner
+ * @param actions callable to execute
+ * @return value returned from actions
+ * @throws Exception
+ */
+ public T ignorePermissions(Callable<T> actions) throws Exception {
+ ignorePermissions = true;
+ T result;
+ try {
+ result = actions.call();
+ } finally {
+ ignorePermissions = false;
+ }
+ return result;
+ }
+
+ protected String getUsername() {
+ return context.getUsername();
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/b88db3cc/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/SharedCRUDResourceManager.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/SharedCRUDResourceManager.java b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/SharedCRUDResourceManager.java
new file mode 100644
index 0000000..3942276
--- /dev/null
+++ b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/SharedCRUDResourceManager.java
@@ -0,0 +1,44 @@
+/**
+ * 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.hive2.resources;
+
+import org.apache.ambari.view.ViewContext;
+import org.apache.ambari.view.hive2.persistence.IStorageFactory;
+import org.apache.ambari.view.hive2.persistence.utils.Indexed;
+
+/**
+ * Resource manager that doesn't restrict access (Allow all)
+ * @param <T> Data type with ID
+ */
+public class SharedCRUDResourceManager<T extends Indexed> extends CRUDResourceManager<T> {
+ protected ViewContext context;
+
+ /**
+ * Constructor
+ * @param responseClass model class
+ */
+ public SharedCRUDResourceManager(Class<T> responseClass, IStorageFactory storageFabric) {
+ super(responseClass, storageFabric);
+ }
+
+ @Override
+ protected boolean checkPermissions(T object) {
+ return true; //everyone has permission
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/b88db3cc/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/browser/HiveBrowserService.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/browser/HiveBrowserService.java b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/browser/HiveBrowserService.java
new file mode 100644
index 0000000..0da49c4
--- /dev/null
+++ b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/browser/HiveBrowserService.java
@@ -0,0 +1,261 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.hive2.resources.browser;
+
+import org.apache.ambari.view.ViewContext;
+import org.apache.ambari.view.ViewResourceHandler;
+import org.apache.ambari.view.hive2.client.ColumnDescription;
+import org.apache.ambari.view.hive2.client.ConnectionConfig;
+import org.apache.ambari.view.hive2.client.Cursor;
+import org.apache.ambari.view.hive2.client.DDLDelegator;
+import org.apache.ambari.view.hive2.client.DDLDelegatorImpl;
+import org.apache.ambari.view.hive2.client.Row;
+import org.apache.ambari.view.hive2.resources.jobs.ResultsPaginationController;
+import org.apache.ambari.view.hive2.utils.BadRequestFormattedException;
+import org.apache.ambari.view.hive2.utils.ServiceFormattedException;
+import org.apache.ambari.view.hive2.ConnectionFactory;
+import org.apache.ambari.view.hive2.ConnectionSystem;
+import org.json.simple.JSONObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.inject.Inject;
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
+import javax.ws.rs.QueryParam;
+import javax.ws.rs.WebApplicationException;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import java.util.List;
+import java.util.concurrent.Callable;
+
+/**
+ * Database access resource
+ */
+public class HiveBrowserService {
+ @Inject
+ ViewResourceHandler handler;
+ @Inject
+ protected ViewContext context;
+
+ protected final static Logger LOG =
+ LoggerFactory.getLogger(HiveBrowserService.class);
+
+ /**
+ * Returns list of databases
+ */
+ @GET
+ @Path("database")
+ @Produces(MediaType.APPLICATION_JSON)
+ public Response databases(@QueryParam("like") String like,
+ @QueryParam("first") String fromBeginning,
+ @QueryParam("count") Integer count,
+ @QueryParam("columns") final String requestedColumns) {
+ if (like == null)
+ like = "*";
+ else
+ like = "*" + like + "*";
+ JSONObject response = new JSONObject();
+ DDLDelegator delegator = new DDLDelegatorImpl(context, ConnectionSystem.getInstance().getActorSystem(), ConnectionSystem.getInstance().getOperationController(context));
+ List<String> databases = delegator.getDbList(getHiveConnectionConfig(), like);
+ response.put("databases", databases);
+ return Response.ok(response).build();
+
+ }
+
+ /**
+ * Returns list of databases
+ */
+ @GET
+ @Path("database.page")
+ @Produces(MediaType.APPLICATION_JSON)
+ public Response databasesPaginated(@QueryParam("like") String like,
+ @QueryParam("first") String fromBeginning,
+ @QueryParam("count") Integer count,
+ @QueryParam("searchId") String searchId,
+ @QueryParam("format") String format,
+ @QueryParam("columns") final String requestedColumns) {
+ if (like == null)
+ like = "*";
+ else
+ like = "*" + like + "*";
+ String curl = null;
+ try {
+ final String finalLike = like;
+ final DDLDelegator delegator = new DDLDelegatorImpl(context, ConnectionSystem.getInstance().getActorSystem(), ConnectionSystem.getInstance().getOperationController(context));
+ return ResultsPaginationController.getInstance(context)
+ .request("databases", searchId, false, fromBeginning, count, format, requestedColumns,
+ new Callable<Cursor<Row, ColumnDescription>>() {
+ @Override
+ public Cursor<Row, ColumnDescription> call() throws Exception {
+ return delegator.getDbListCursor(getHiveConnectionConfig(), finalLike);
+ }
+ }).build();
+ } catch (WebApplicationException ex) {
+ throw ex;
+ } catch (IllegalArgumentException ex) {
+ throw new BadRequestFormattedException(ex.getMessage(), ex);
+ } catch (Exception ex) {
+ throw new ServiceFormattedException(ex.getMessage(), ex, curl);
+ }
+ }
+
+ /**
+ * Returns list of databases
+ */
+ @GET
+ @Path("database/{db}/table")
+ @Produces(MediaType.APPLICATION_JSON)
+ public Response tablesInDatabase(@PathParam("db") String db,
+ @QueryParam("like") String like,
+ @QueryParam("first") String fromBeginning,
+ @QueryParam("count") Integer count,
+ @QueryParam("columns") final String requestedColumns) {
+ if (like == null)
+ like = "*";
+ else
+ like = "*" + like + "*";
+
+ JSONObject response = new JSONObject();
+ DDLDelegator delegator = new DDLDelegatorImpl(context, ConnectionSystem.getInstance().getActorSystem(), ConnectionSystem.getInstance().getOperationController(context));
+ List<String> tables = delegator.getTableList(getHiveConnectionConfig(), db, like);
+ response.put("tables", tables);
+ response.put("database", db);
+ return Response.ok(response).build();
+
+ }
+
+ /**
+ * Returns list of databases
+ */
+ @GET
+ @Path("database/{db}/table.page")
+ @Produces(MediaType.APPLICATION_JSON)
+ public Response tablesInDatabasePaginated(@PathParam("db") final String db,
+ @QueryParam("like") String like,
+ @QueryParam("first") String fromBeginning,
+ @QueryParam("count") Integer count,
+ @QueryParam("searchId") String searchId,
+ @QueryParam("format") String format,
+ @QueryParam("columns") final String requestedColumns) {
+ if (like == null)
+ like = "*";
+ else
+ like = "*" + like + "*";
+ String curl = null;
+ try {
+ final String finalLike = like;
+ final DDLDelegator delegator = new DDLDelegatorImpl(context, ConnectionSystem.getInstance().getActorSystem(), ConnectionSystem.getInstance().getOperationController(context));
+ try {
+ return ResultsPaginationController.getInstance(context)
+ .request(db + ":tables:", searchId, false, fromBeginning, count, format, requestedColumns,
+ new Callable<Cursor<Row, ColumnDescription>>() {
+ @Override
+ public Cursor<Row, ColumnDescription> call() throws Exception {
+ return delegator.getTableListCursor(getHiveConnectionConfig(), db, finalLike);
+ }
+ }).build();
+ } catch (Exception ex) {
+ throw new ServiceFormattedException(ex.getMessage(), ex);
+ }
+
+ } catch (WebApplicationException ex) {
+ throw ex;
+ } catch (IllegalArgumentException ex) {
+ throw new BadRequestFormattedException(ex.getMessage(), ex);
+ } catch (Exception ex) {
+ throw new ServiceFormattedException(ex.getMessage(), ex, curl);
+ }
+ }
+
+ /**
+ * Returns list of databases
+ */
+ @GET
+ @Path("database/{db}/table/{table}")
+ @Produces(MediaType.APPLICATION_JSON)
+ public Response describeTable(@PathParam("db") String db,
+ @PathParam("table") String table,
+ @QueryParam("like") String like,
+ @QueryParam("columns") String requestedColumns,
+ @QueryParam("extended") String extended) {
+ boolean extendedTableDescription = (extended != null && extended.equals("true"));
+ String curl = null;
+ try {
+ JSONObject response = new JSONObject();
+ DDLDelegator delegator = new DDLDelegatorImpl(context, ConnectionSystem.getInstance().getActorSystem(), ConnectionSystem.getInstance().getOperationController(context));
+ List<ColumnDescription> descriptions = delegator.getTableDescription(getHiveConnectionConfig(), db, table, "%", extendedTableDescription);
+ response.put("columns", descriptions);
+ response.put("database", db);
+ response.put("table", table);
+
+ //TODO: New implementation
+
+ return Response.ok(response).build();
+ } catch (WebApplicationException ex) {
+ throw ex;
+ } catch (IllegalArgumentException ex) {
+ throw new BadRequestFormattedException(ex.getMessage(), ex);
+ } catch (Exception ex) {
+ throw new ServiceFormattedException(ex.getMessage(), ex, curl);
+ }
+ }
+
+ /**
+ * Returns list of databases
+ */
+ @GET
+ @Path("database/{db}/table/{table}.page")
+ @Produces(MediaType.APPLICATION_JSON)
+ public Response describeTablePaginated(@PathParam("db") final String db,
+ @PathParam("table") final String table,
+ @QueryParam("like") String like,
+ @QueryParam("first") String fromBeginning,
+ @QueryParam("searchId") String searchId,
+ @QueryParam("count") Integer count,
+ @QueryParam("format") String format,
+ @QueryParam("columns") final String requestedColumns) {
+ if (like == null)
+ like = ".*";
+ else
+ like = ".*" + like + ".*";
+ final String finalLike = like;
+
+ final DDLDelegator delegator = new DDLDelegatorImpl(context, ConnectionSystem.getInstance().getActorSystem(), ConnectionSystem.getInstance().getOperationController(context));
+ try {
+ return ResultsPaginationController.getInstance(context)
+ .request(db + ":tables:" + table + ":columns", searchId, false, fromBeginning, count, format, requestedColumns,
+ new Callable<Cursor<Row, ColumnDescription>>() {
+ @Override
+ public Cursor<Row, ColumnDescription> call() throws Exception {
+ return delegator.getTableDescriptionCursor(getHiveConnectionConfig(), db, table, finalLike, false);
+ }
+ }).build();
+ } catch (Exception ex) {
+ throw new ServiceFormattedException(ex.getMessage(), ex);
+ }
+ }
+
+
+ private ConnectionConfig getHiveConnectionConfig() {
+ return ConnectionFactory.create(context);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/b88db3cc/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/files/FileResource.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/files/FileResource.java b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/files/FileResource.java
new file mode 100644
index 0000000..e8dfce9
--- /dev/null
+++ b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/files/FileResource.java
@@ -0,0 +1,70 @@
+/**
+ * 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.hive2.resources.files;
+
+/**
+ * File bean
+ */
+public class FileResource {
+ private String filePath;
+ private String fileContent;
+ private boolean hasNext;
+ private long page;
+ private long pageCount;
+
+ public String getFilePath() {
+ return filePath;
+ }
+
+ public void setFilePath(String filePath) {
+ this.filePath = filePath;
+ }
+
+ public String getFileContent() {
+ return fileContent;
+ }
+
+ public void setFileContent(String fileContent) {
+ this.fileContent = fileContent;
+ }
+
+ public boolean isHasNext() {
+ return hasNext;
+ }
+
+ public void setHasNext(boolean hasNext) {
+ this.hasNext = hasNext;
+ }
+
+ public long getPage() {
+ return page;
+ }
+
+ public void setPage(long page) {
+ this.page = page;
+ }
+
+ public long getPageCount() {
+ return pageCount;
+ }
+
+ public void setPageCount(long pageCount) {
+ this.pageCount = pageCount;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ambari/blob/b88db3cc/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/files/FileService.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/files/FileService.java b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/files/FileService.java
new file mode 100644
index 0000000..db63364
--- /dev/null
+++ b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/files/FileService.java
@@ -0,0 +1,250 @@
+/**
+ * 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.hive2.resources.files;
+
+import com.jayway.jsonpath.JsonPath;
+import org.apache.ambari.view.ViewContext;
+import org.apache.ambari.view.ViewResourceHandler;
+import org.apache.ambari.view.hive2.BaseService;
+import org.apache.ambari.view.hive2.utils.*;
+import org.apache.ambari.view.utils.hdfs.HdfsApi;
+import org.apache.ambari.view.utils.hdfs.HdfsUtil;
+import org.apache.commons.codec.binary.Base64;
+import org.apache.commons.io.IOUtils;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileAlreadyExistsException;
+import org.json.simple.JSONObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.inject.Inject;
+import javax.servlet.http.HttpServletResponse;
+import javax.ws.rs.*;
+import javax.ws.rs.core.Context;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import javax.ws.rs.core.UriInfo;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URL;
+import java.util.HashMap;
+
+/**
+ * File access resource
+ * API:
+ * GET /:path
+ * read entire file
+ * POST /
+ * create new file
+ * Required: filePath
+ * file should not already exists
+ * PUT /:path
+ * update file content
+ */
+public class FileService extends BaseService {
+ public static final String FAKE_FILE = "fakefile://";
+ public static final String JSON_PATH_FILE = "jsonpath:";
+
+ @Inject
+ ViewResourceHandler handler;
+
+ protected final static Logger LOG =
+ LoggerFactory.getLogger(FileService.class);
+
+ /**
+ * Get single item
+ */
+ @GET
+ @Path("{filePath:.*}")
+ @Produces(MediaType.APPLICATION_JSON)
+ public Response getFilePage(@PathParam("filePath") String filePath, @QueryParam("page") Long page) throws IOException, InterruptedException {
+
+ LOG.debug("Reading file " + filePath);
+ try {
+ FileResource file = new FileResource();
+
+ if (page == null)
+ page = 0L;
+
+ if (filePath.startsWith(FAKE_FILE)) {
+ if (page > 1)
+ throw new IllegalArgumentException("There's only one page in fake files");
+
+ String encodedContent = filePath.substring(FAKE_FILE.length());
+ String content = new String(Base64.decodeBase64(encodedContent));
+
+ fillFakeFileObject(filePath, file, content);
+ } else if (filePath.startsWith(JSON_PATH_FILE)) {
+ if (page > 1)
+ throw new IllegalArgumentException("There's only one page in fake files");
+
+ String content = getJsonPathContentByUrl(filePath);
+ fillFakeFileObject(filePath, file, content);
+ } else {
+
+ filePath = sanitizeFilePath(filePath);
+ FilePaginator paginator = new FilePaginator(filePath, getSharedObjectsFactory().getHdfsApi());
+
+ fillRealFileObject(filePath, page, file, paginator);
+ }
+
+ JSONObject object = new JSONObject();
+ object.put("file", file);
+ return Response.ok(object).status(200).build();
+ } catch (WebApplicationException ex) {
+ throw ex;
+ } catch (FileNotFoundException ex) {
+ throw new NotFoundFormattedException(ex.getMessage(), ex);
+ } catch (IllegalArgumentException ex) {
+ throw new BadRequestFormattedException(ex.getMessage(), ex);
+ } catch (Exception ex) {
+ throw new ServiceFormattedException(ex.getMessage(), ex);
+ }
+ }
+
+ protected String getJsonPathContentByUrl(String filePath) throws IOException {
+ URL url = new URL(filePath.substring(JSON_PATH_FILE.length()));
+
+ InputStream responseInputStream = context.getURLStreamProvider().readFrom(url.toString(), "GET",
+ (String)null, new HashMap<String, String>());
+ String response = IOUtils.toString(responseInputStream);
+
+ for (String ref : url.getRef().split("!")) {
+ response = JsonPath.read(response, ref);
+ }
+ return response;
+ }
+
+ public void fillRealFileObject(String filePath, Long page, FileResource file, FilePaginator paginator) throws IOException, InterruptedException {
+ file.setFilePath(filePath);
+ file.setFileContent(paginator.readPage(page));
+ file.setHasNext(paginator.pageCount() > page + 1);
+ file.setPage(page);
+ file.setPageCount(paginator.pageCount());
+ }
+
+ public void fillFakeFileObject(String filePath, FileResource file, String content) {
+ file.setFilePath(filePath);
+ file.setFileContent(content);
+ file.setHasNext(false);
+ file.setPage(0);
+ file.setPageCount(1);
+ }
+
+ /**
+ * Delete single item
+ */
+ @DELETE
+ @Path("{filePath:.*}")
+ public Response deleteFile(@PathParam("filePath") String filePath) throws IOException, InterruptedException {
+ try {
+ filePath = sanitizeFilePath(filePath);
+ LOG.debug("Deleting file " + filePath);
+ if (getSharedObjectsFactory().getHdfsApi().delete(filePath, false)) {
+ return Response.status(204).build();
+ }
+ throw new NotFoundFormattedException("FileSystem.delete returned false", null);
+ } catch (WebApplicationException ex) {
+ throw ex;
+ } catch (Exception ex) {
+ throw new ServiceFormattedException(ex.getMessage(), ex);
+ }
+ }
+
+ /**
+ * Update item
+ */
+ @PUT
+ @Path("{filePath:.*}")
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response updateFile(FileResourceRequest request,
+ @PathParam("filePath") String filePath) throws IOException, InterruptedException {
+ try {
+ filePath = sanitizeFilePath(filePath);
+ LOG.debug("Rewriting file " + filePath);
+ FSDataOutputStream output = getSharedObjectsFactory().getHdfsApi().create(filePath, true);
+ output.writeBytes(request.file.getFileContent());
+ output.close();
+ return Response.status(204).build();
+ } catch (WebApplicationException ex) {
+ throw ex;
+ } catch (Exception ex) {
+ throw new ServiceFormattedException(ex.getMessage(), ex);
+ }
+ }
+
+ /**
+ * Create script
+ */
+ @POST
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response createFile(FileResourceRequest request,
+ @Context HttpServletResponse response, @Context UriInfo ui)
+ throws IOException, InterruptedException {
+ try {
+ LOG.debug("Creating file " + request.file.getFilePath());
+ try {
+ FSDataOutputStream output = getSharedObjectsFactory().getHdfsApi().create(request.file.getFilePath(), false);
+ if (request.file.getFileContent() != null) {
+ output.writeBytes(request.file.getFileContent());
+ }
+ output.close();
+ } catch (FileAlreadyExistsException ex) {
+ throw new ServiceFormattedException("F020 File already exists", ex, 400);
+ }
+ response.setHeader("Location",
+ String.format("%s/%s", ui.getAbsolutePath().toString(), request.file.getFilePath()));
+ return Response.status(204).build();
+ } catch (WebApplicationException ex) {
+ throw ex;
+ } catch (Exception ex) {
+ throw new ServiceFormattedException(ex.getMessage(), ex);
+ }
+ }
+
+ /**
+ * Checks connection to HDFS
+ * @param context View Context
+ */
+ public static void hdfsSmokeTest(ViewContext context) {
+ try {
+ HdfsApi api = HdfsUtil.connectToHDFSApi(context);
+ api.getStatus();
+ } catch (WebApplicationException ex) {
+ throw ex;
+ } catch (Exception ex) {
+ throw new ServiceFormattedException(ex.getMessage(), ex);
+ }
+ }
+
+ /**
+ * Wrapper object for json mapping
+ */
+ public static class FileResourceRequest {
+ public FileResource file;
+ }
+
+ private String sanitizeFilePath(String filePath){
+ if (!filePath.startsWith("/") && !filePath.startsWith(".")) {
+ filePath = "/" + filePath; // some servers strip double slashes in URL
+ }
+ return filePath;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/b88db3cc/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/Aggregator.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/Aggregator.java b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/Aggregator.java
new file mode 100644
index 0000000..e7553dd
--- /dev/null
+++ b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/Aggregator.java
@@ -0,0 +1,231 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.hive2.resources.jobs;
+
+import org.apache.ambari.view.hive2.persistence.utils.FilteringStrategy;
+import org.apache.ambari.view.hive2.persistence.utils.Indexed;
+import org.apache.ambari.view.hive2.persistence.utils.ItemNotFound;
+import org.apache.ambari.view.hive2.resources.IResourceManager;
+import org.apache.ambari.view.hive2.resources.files.FileService;
+import org.apache.ambari.view.hive2.resources.jobs.atsJobs.HiveQueryId;
+import org.apache.ambari.view.hive2.resources.jobs.atsJobs.IATSParser;
+import org.apache.ambari.view.hive2.resources.jobs.atsJobs.TezDagId;
+import org.apache.ambari.view.hive2.resources.jobs.viewJobs.Job;
+import org.apache.ambari.view.hive2.resources.jobs.viewJobs.JobImpl;
+import org.apache.commons.beanutils.PropertyUtils;
+import org.apache.commons.codec.binary.Base64;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * View Jobs and ATS Jobs aggregator.
+ * There are 4 options:
+ * 1) ATS ExecuteJob without operationId
+ * *Meaning*: executed outside of HS2
+ * - ExecuteJob info only from ATS
+ * 2) ATS ExecuteJob with operationId
+ * a) Hive View ExecuteJob with same operationId is not present
+ * *Meaning*: executed with HS2
+ * - ExecuteJob info only from ATS
+ * b) Hive View ExecuteJob with operationId is present (need to merge)
+ * *Meaning*: executed with HS2 through Hive View
+ * - ExecuteJob info merged from ATS and from Hive View DataStorage
+ * 3) ExecuteJob present only in Hive View, ATS does not have it
+ * *Meaning*: executed through Hive View, but Hadoop ExecuteJob was not created
+ * it can happen if user executes query without aggregation, like just "select * from TABLE"
+ * - ExecuteJob info only from Hive View
+ */
+public class Aggregator {
+ protected final static Logger LOG =
+ LoggerFactory.getLogger(Aggregator.class);
+
+ private final IATSParser ats;
+ private IResourceManager<Job> viewJobResourceManager;
+
+ public Aggregator(IResourceManager<Job> jobResourceManager,
+ IATSParser ats) {
+ this.viewJobResourceManager = jobResourceManager;
+ this.ats = ats;
+ }
+
+ public List<Job> readAll(String username) {
+ Set<String> addedOperationIds = new HashSet<>();
+
+ List<Job> allJobs = new LinkedList<>();
+ for (HiveQueryId atsHiveQuery : ats.getHiveQueryIdsList(username)) {
+
+ TezDagId atsTezDag = getTezDagFromHiveQueryId(atsHiveQuery);
+
+ JobImpl atsJob;
+ if (hasOperationId(atsHiveQuery)) {
+ try {
+ Job viewJob = getJobByOperationId(urlSafeBase64ToHexString(atsHiveQuery.operationId));
+ saveJobInfoIfNeeded(atsHiveQuery, atsTezDag, viewJob);
+
+ atsJob = mergeAtsJobWithViewJob(atsHiveQuery, atsTezDag, viewJob);
+ } catch (ItemNotFound itemNotFound) {
+ // Executed from HS2, but outside of Hive View
+ atsJob = atsOnlyJob(atsHiveQuery, atsTezDag);
+ }
+ } else {
+ atsJob = atsOnlyJob(atsHiveQuery, atsTezDag);
+ }
+ allJobs.add(atsJob);
+
+ addedOperationIds.add(atsHiveQuery.operationId);
+ }
+
+ return allJobs;
+ }
+
+ public Job readATSJob(Job viewJob) throws ItemNotFound {
+
+ if (viewJob.getStatus().equals(Job.JOB_STATE_INITIALIZED) || viewJob.getStatus().equals(Job.JOB_STATE_UNKNOWN))
+ return viewJob;
+
+ String hexGuid = viewJob.getGuid();
+
+
+ HiveQueryId atsHiveQuery = ats.getHiveQueryIdByOperationId(hexGuid);
+
+ TezDagId atsTezDag = getTezDagFromHiveQueryId(atsHiveQuery);
+
+ saveJobInfoIfNeeded(atsHiveQuery, atsTezDag, viewJob);
+ return mergeAtsJobWithViewJob(atsHiveQuery, atsTezDag, viewJob);
+ }
+
+ private TezDagId getTezDagFromHiveQueryId(HiveQueryId atsHiveQuery) {
+ TezDagId atsTezDag;
+ if (atsHiveQuery.version >= HiveQueryId.ATS_15_RESPONSE_VERSION) {
+ atsTezDag = ats.getTezDAGByEntity(atsHiveQuery.entity);
+ } else if (atsHiveQuery.dagNames != null && atsHiveQuery.dagNames.size() > 0) {
+ String dagName = atsHiveQuery.dagNames.get(0);
+
+ atsTezDag = ats.getTezDAGByName(dagName);
+ } else {
+ atsTezDag = new TezDagId();
+ }
+ return atsTezDag;
+ }
+
+ protected boolean hasOperationId(HiveQueryId atsHiveQuery) {
+ return atsHiveQuery.operationId != null;
+ }
+
+ protected JobImpl mergeAtsJobWithViewJob(HiveQueryId atsHiveQuery, TezDagId atsTezDag, Job viewJob) {
+ JobImpl atsJob;
+ try {
+ atsJob = new JobImpl(PropertyUtils.describe(viewJob));
+ } catch (IllegalAccessException e) {
+ LOG.error("Can't instantiate JobImpl", e);
+ return null;
+ } catch (InvocationTargetException e) {
+ LOG.error("Can't instantiate JobImpl", e);
+ return null;
+ } catch (NoSuchMethodException e) {
+ LOG.error("Can't instantiate JobImpl", e);
+ return null;
+ }
+ fillAtsJobFields(atsJob, atsHiveQuery, atsTezDag);
+ return atsJob;
+ }
+
+ protected void saveJobInfoIfNeeded(HiveQueryId hiveQueryId, TezDagId tezDagId, Job viewJob) throws ItemNotFound {
+ if (viewJob.getDagName() == null || viewJob.getDagName().isEmpty()) {
+ if (hiveQueryId.dagNames != null && hiveQueryId.dagNames.size() > 0) {
+ viewJob.setDagName(hiveQueryId.dagNames.get(0));
+ viewJobResourceManager.update(viewJob, viewJob.getId());
+ }
+ }
+ if (tezDagId.status != null && (tezDagId.status.compareToIgnoreCase(Job.JOB_STATE_UNKNOWN) != 0) &&
+ !viewJob.getStatus().equalsIgnoreCase(tezDagId.status)) {
+ viewJob.setDagId(tezDagId.entity);
+ viewJob.setApplicationId(tezDagId.applicationId);
+ if (!(viewJob.getStatus().equalsIgnoreCase(Job.JOB_STATE_FINISHED)
+ || (viewJob.getStatus().equalsIgnoreCase(Job.JOB_STATE_ERROR))
+ || viewJob.getStatus().equalsIgnoreCase(Job.JOB_STATE_CANCELED))) {
+ viewJob.setStatus(tezDagId.status);
+ }
+ viewJobResourceManager.update(viewJob, viewJob.getId());
+ }
+ }
+
+ protected JobImpl atsOnlyJob(HiveQueryId atsHiveQuery, TezDagId atsTezDag) {
+ JobImpl atsJob = new JobImpl();
+ atsJob.setId(atsHiveQuery.entity);
+ fillAtsJobFields(atsJob, atsHiveQuery, atsTezDag);
+
+ String query = atsHiveQuery.query;
+ atsJob.setTitle(query.substring(0, (query.length() > 42) ? 42 : query.length()));
+
+ atsJob.setQueryFile(FileService.JSON_PATH_FILE + atsHiveQuery.url + "#otherinfo.QUERY!queryText");
+ return atsJob;
+ }
+
+ protected JobImpl fillAtsJobFields(JobImpl atsJob, HiveQueryId atsHiveQuery, TezDagId atsTezDag) {
+ atsJob.setApplicationId(atsTezDag.applicationId);
+
+ if (atsHiveQuery.dagNames != null && atsHiveQuery.dagNames.size() > 0)
+ atsJob.setDagName(atsHiveQuery.dagNames.get(0));
+ atsJob.setDagId(atsTezDag.entity);
+ if (atsTezDag.status != null && !atsTezDag.status.equals(TezDagId.STATUS_UNKNOWN))
+ atsJob.setStatus(atsTezDag.status);
+ if (atsHiveQuery.starttime != 0)
+ atsJob.setDateSubmitted(atsHiveQuery.starttime);
+ atsJob.setDuration(atsHiveQuery.duration);
+ return atsJob;
+ }
+
+ protected Job getJobByOperationId(final String opId) throws ItemNotFound {
+ List<Job> operationHandles = viewJobResourceManager.readAll(new FilteringStrategy() {
+ @Override
+ public boolean isConform(Indexed item) {
+ Job opHandle = (Job) item;
+ return opHandle.getGuid().equals(opId);
+ }
+
+ @Override
+ public String whereStatement() {
+ return "guid='" + opId + "'";
+ }
+ });
+
+ if (operationHandles.size() != 1)
+ throw new ItemNotFound();
+
+ return viewJobResourceManager.read(operationHandles.get(0).getId());
+ }
+
+ protected static String urlSafeBase64ToHexString(String urlsafeBase64) {
+ byte[] decoded = Base64.decodeBase64(urlsafeBase64);
+
+ StringBuilder sb = new StringBuilder();
+ for (byte b : decoded) {
+ sb.append(String.format("%02x", b));
+ }
+ return sb.toString();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/b88db3cc/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/JobResourceProvider.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/JobResourceProvider.java b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/JobResourceProvider.java
new file mode 100644
index 0000000..28573b3
--- /dev/null
+++ b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/JobResourceProvider.java
@@ -0,0 +1,117 @@
+/**
+ * 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.hive2.resources.jobs;
+
+import org.apache.ambari.view.*;
+import org.apache.ambari.view.hive2.persistence.utils.ItemNotFound;
+import org.apache.ambari.view.hive2.persistence.utils.OnlyOwnersFilteringStrategy;
+import org.apache.ambari.view.hive2.resources.jobs.viewJobs.*;
+import org.apache.ambari.view.hive2.utils.SharedObjectsFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.inject.Inject;
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Resource provider for job
+ */
+public class JobResourceProvider implements ResourceProvider<Job> {
+ @Inject
+ ViewContext context;
+
+ protected JobResourceManager resourceManager = null;
+ protected final static Logger LOG =
+ LoggerFactory.getLogger(JobResourceProvider.class);
+
+ protected synchronized JobResourceManager getResourceManager() {
+ if (resourceManager == null) {
+ resourceManager = new JobResourceManager(new SharedObjectsFactory(context), context);
+ }
+ return resourceManager;
+ }
+
+ @Override
+ public Job getResource(String resourceId, Set<String> properties) throws SystemException, NoSuchResourceException, UnsupportedPropertyException {
+ try {
+ return getResourceManager().read(resourceId);
+ } catch (ItemNotFound itemNotFound) {
+ throw new NoSuchResourceException(resourceId);
+ }
+ }
+
+ @Override
+ public Set<Job> getResources(ReadRequest readRequest) throws SystemException, NoSuchResourceException, UnsupportedPropertyException {
+ if (context == null) {
+ return new HashSet<Job>();
+ }
+ return new HashSet<Job>(getResourceManager().readAll(
+ new OnlyOwnersFilteringStrategy(this.context.getUsername())));
+ }
+
+ @Override
+ public void createResource(String s, Map<String, Object> stringObjectMap) throws SystemException, ResourceAlreadyExistsException, NoSuchResourceException, UnsupportedPropertyException {
+ Job item = null;
+ try {
+ item = new JobImpl(stringObjectMap);
+ } catch (InvocationTargetException e) {
+ throw new SystemException("error on creating resource", e);
+ } catch (IllegalAccessException e) {
+ throw new SystemException("error on creating resource", e);
+ }
+ getResourceManager().create(item);
+ JobController jobController = new SharedObjectsFactory(context).getJobControllerFactory().createControllerForJob(item);
+ try {
+ jobController.submit();
+ } catch (Throwable throwable) {
+ throw new SystemException("error on creating resource", throwable);
+ }
+ }
+
+ @Override
+ public boolean updateResource(String resourceId, Map<String, Object> stringObjectMap) throws SystemException, NoSuchResourceException, UnsupportedPropertyException {
+ Job item = null;
+ try {
+ item = new JobImpl(stringObjectMap);
+ } catch (InvocationTargetException e) {
+ throw new SystemException("error on updating resource", e);
+ } catch (IllegalAccessException e) {
+ throw new SystemException("error on updating resource", e);
+ }
+ try {
+ getResourceManager().update(item, resourceId);
+ } catch (ItemNotFound itemNotFound) {
+ throw new NoSuchResourceException(resourceId);
+ }
+ return true;
+ }
+
+ @Override
+ public boolean deleteResource(String resourceId) throws SystemException, NoSuchResourceException, UnsupportedPropertyException {
+ try {
+ getResourceManager().delete(resourceId);
+ } catch (ItemNotFound itemNotFound) {
+ throw new NoSuchResourceException(resourceId);
+ }
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/b88db3cc/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/JobService.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/JobService.java b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/JobService.java
new file mode 100644
index 0000000..4fa7f2a
--- /dev/null
+++ b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/JobService.java
@@ -0,0 +1,612 @@
+/**
+ * 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.hive2.resources.jobs;
+
+import com.beust.jcommander.internal.Lists;
+import com.google.common.base.Optional;
+import org.apache.ambari.view.ViewResourceHandler;
+import org.apache.ambari.view.hive2.BaseService;
+import org.apache.ambari.view.hive2.actor.message.job.AsyncExecutionFailed;
+import org.apache.ambari.view.hive2.backgroundjobs.BackgroundJobController;
+import org.apache.ambari.view.hive2.client.AsyncJobRunner;
+import org.apache.ambari.view.hive2.client.AsyncJobRunnerImpl;
+import org.apache.ambari.view.hive2.client.ColumnDescription;
+import org.apache.ambari.view.hive2.client.Cursor;
+import org.apache.ambari.view.hive2.client.EmptyCursor;
+import org.apache.ambari.view.hive2.client.HiveClientException;
+import org.apache.ambari.view.hive2.client.NonPersistentCursor;
+import org.apache.ambari.view.hive2.client.Row;
+import org.apache.ambari.view.hive2.persistence.utils.ItemNotFound;
+import org.apache.ambari.view.hive2.resources.jobs.atsJobs.IATSParser;
+import org.apache.ambari.view.hive2.resources.jobs.viewJobs.Job;
+import org.apache.ambari.view.hive2.resources.jobs.viewJobs.JobController;
+import org.apache.ambari.view.hive2.resources.jobs.viewJobs.JobImpl;
+import org.apache.ambari.view.hive2.resources.jobs.viewJobs.JobResourceManager;
+import org.apache.ambari.view.hive2.utils.MisconfigurationFormattedException;
+import org.apache.ambari.view.hive2.utils.NotFoundFormattedException;
+import org.apache.ambari.view.hive2.utils.ServiceFormattedException;
+import org.apache.ambari.view.hive2.utils.SharedObjectsFactory;
+import org.apache.ambari.view.hive2.ConnectionSystem;
+import org.apache.commons.beanutils.PropertyUtils;
+import org.apache.commons.csv.CSVFormat;
+import org.apache.commons.csv.CSVPrinter;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.json.simple.JSONObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.inject.Inject;
+import javax.servlet.http.HttpServletResponse;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.DELETE;
+import javax.ws.rs.GET;
+import javax.ws.rs.POST;
+import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
+import javax.ws.rs.QueryParam;
+import javax.ws.rs.WebApplicationException;
+import javax.ws.rs.core.Context;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import javax.ws.rs.core.StreamingOutput;
+import javax.ws.rs.core.UriInfo;
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.io.OutputStreamWriter;
+import java.io.Writer;
+import java.lang.reflect.InvocationTargetException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+/**
+ * Servlet for queries
+ * API:
+ * GET /:id
+ * read job
+ * POST /
+ * create new job
+ * Required: title, queryFile
+ * GET /
+ * get all Jobs of current user
+ */
+public class JobService extends BaseService {
+ @Inject
+ ViewResourceHandler handler;
+
+ private JobResourceManager resourceManager;
+
+ protected final static Logger LOG =
+ LoggerFactory.getLogger(JobService.class);
+ private Aggregator aggregator;
+
+ protected synchronized JobResourceManager getResourceManager() {
+ if (resourceManager == null) {
+ SharedObjectsFactory connectionsFactory = getSharedObjectsFactory();
+ resourceManager = new JobResourceManager(connectionsFactory, context);
+ }
+ return resourceManager;
+ }
+
+
+ protected Aggregator getAggregator() {
+ if (aggregator == null) {
+ IATSParser atsParser = getSharedObjectsFactory().getATSParser();
+ aggregator = new Aggregator(getResourceManager(), atsParser);
+ }
+ return aggregator;
+ }
+
+ protected void setAggregator(Aggregator aggregator) {
+ this.aggregator = aggregator;
+ }
+
+ /**
+ * Get single item
+ */
+ @GET
+ @Path("{jobId}")
+ @Produces(MediaType.APPLICATION_JSON)
+ public Response getOne(@PathParam("jobId") String jobId) {
+ try {
+ JobController jobController = getResourceManager().readController(jobId);
+
+ Job job = jobController.getJob();
+ if(job.getStatus().equals(Job.JOB_STATE_ERROR)){
+ ConnectionSystem system = ConnectionSystem.getInstance();
+ final AsyncJobRunner asyncJobRunner = new AsyncJobRunnerImpl(context, system.getOperationController(context), system.getActorSystem());
+ Optional<AsyncExecutionFailed> error = asyncJobRunner.getError(jobId, context.getUsername());
+ if(error.isPresent()){
+ throw new Exception(error.get().getError());
+ }
+ }
+
+ JSONObject jsonJob = jsonObjectFromJob(jobController);
+ return Response.ok(jsonJob).build();
+ } catch (WebApplicationException ex) {
+ throw ex;
+ } catch (ItemNotFound itemNotFound) {
+ throw new NotFoundFormattedException(itemNotFound.getMessage(), itemNotFound);
+ } catch (Exception ex) {
+ throw new ServiceFormattedException(ex.getMessage(), ex);
+ }
+ }
+
+ private JSONObject jsonObjectFromJob(JobController jobController) throws IllegalAccessException, NoSuchMethodException, InvocationTargetException {
+ Job hiveJob = jobController.getJobPOJO();
+
+ Job mergedJob;
+ try {
+ mergedJob = getAggregator().readATSJob(hiveJob);
+ } catch (ItemNotFound itemNotFound) {
+ throw new ServiceFormattedException("E010 ExecuteJob not found", itemNotFound);
+ }
+ Map createdJobMap = PropertyUtils.describe(mergedJob);
+ createdJobMap.remove("class"); // no need to show Bean class on client
+
+ JSONObject jobJson = new JSONObject();
+ jobJson.put("job", createdJobMap);
+ return jobJson;
+ }
+
+ /**
+ * Get job results in csv format
+ */
+ @GET
+ @Path("{jobId}/results/csv")
+ @Produces("text/csv")
+ public Response getResultsCSV(@PathParam("jobId") String jobId,
+ @Context HttpServletResponse response,
+ @QueryParam("fileName") String fileName,
+ @QueryParam("columns") final String requestedColumns) {
+ try {
+
+ final String username = context.getUsername();
+
+ ConnectionSystem system = ConnectionSystem.getInstance();
+ final AsyncJobRunner asyncJobRunner = new AsyncJobRunnerImpl(context, system.getOperationController(context), system.getActorSystem());
+
+ Optional<NonPersistentCursor> cursorOptional = asyncJobRunner.resetAndGetCursor(jobId, username);
+
+ if(!cursorOptional.isPresent()){
+ throw new Exception("Download failed");
+ }
+
+ final NonPersistentCursor resultSet = cursorOptional.get();
+
+
+ StreamingOutput stream = new StreamingOutput() {
+ @Override
+ public void write(OutputStream os) throws IOException, WebApplicationException {
+ Writer writer = new BufferedWriter(new OutputStreamWriter(os));
+ CSVPrinter csvPrinter = new CSVPrinter(writer, CSVFormat.DEFAULT);
+ try {
+
+ List<ColumnDescription> descriptions = resultSet.getDescriptions();
+ List<String> headers = Lists.newArrayList();
+ for (ColumnDescription description : descriptions) {
+ headers.add(description.getName());
+ }
+
+ csvPrinter.printRecord(headers.toArray());
+
+ while (resultSet.hasNext()) {
+ csvPrinter.printRecord(resultSet.next().getRow());
+ writer.flush();
+ }
+ } finally {
+ writer.close();
+ }
+ }
+ };
+
+ if (fileName == null || fileName.isEmpty()) {
+ fileName = "results.csv";
+ }
+
+ return Response.ok(stream).
+ header("Content-Disposition", String.format("attachment; filename=\"%s\"", fileName)).
+ build();
+
+
+ } catch (WebApplicationException ex) {
+ throw ex;
+ } catch (Throwable ex) {
+ throw new ServiceFormattedException(ex.getMessage(), ex);
+ }
+ }
+
+ /**
+ * Get job results in csv format
+ */
+ @GET
+ @Path("{jobId}/results/csv/saveToHDFS")
+ @Produces(MediaType.APPLICATION_JSON)
+ public Response getResultsToHDFS(@PathParam("jobId") String jobId,
+ @QueryParam("commence") String commence,
+ @QueryParam("file") final String targetFile,
+ @QueryParam("stop") final String stop,
+ @QueryParam("columns") final String requestedColumns,
+ @Context HttpServletResponse response) {
+ try {
+
+ final JobController jobController = getResourceManager().readController(jobId);
+ final String username = context.getUsername();
+
+ String backgroundJobId = "csv" + String.valueOf(jobController.getJob().getId());
+ if (commence != null && commence.equals("true")) {
+ if (targetFile == null)
+ throw new MisconfigurationFormattedException("targetFile should not be empty");
+
+ ConnectionSystem system = ConnectionSystem.getInstance();
+ final AsyncJobRunner asyncJobRunner = new AsyncJobRunnerImpl(context, system.getOperationController(context), system.getActorSystem());
+
+ Optional<NonPersistentCursor> cursorOptional = asyncJobRunner.resetAndGetCursor(jobId, username);
+
+ if(!cursorOptional.isPresent()){
+ throw new Exception("Download failed");
+ }
+
+ final NonPersistentCursor resultSet = cursorOptional.get();
+
+ BackgroundJobController.getInstance(context).startJob(String.valueOf(backgroundJobId), new Runnable() {
+ @Override
+ public void run() {
+
+ try {
+
+ FSDataOutputStream stream = getSharedObjectsFactory().getHdfsApi().create(targetFile, true);
+ Writer writer = new BufferedWriter(new OutputStreamWriter(stream));
+ CSVPrinter csvPrinter = new CSVPrinter(writer, CSVFormat.DEFAULT);
+ try {
+ while (resultSet.hasNext() && !Thread.currentThread().isInterrupted()) {
+ csvPrinter.printRecord(resultSet.next().getRow());
+ writer.flush();
+ }
+ } finally {
+ writer.close();
+ }
+ stream.close();
+
+ } catch (IOException e) {
+ throw new ServiceFormattedException("F010 Could not write CSV to HDFS for job#" + jobController.getJob().getId(), e);
+ } catch (InterruptedException e) {
+ throw new ServiceFormattedException("F010 Could not write CSV to HDFS for job#" + jobController.getJob().getId(), e);
+ }
+ }
+ });
+ }
+
+ if (stop != null && stop.equals("true")) {
+ BackgroundJobController.getInstance(context).interrupt(backgroundJobId);
+ }
+
+ JSONObject object = new JSONObject();
+ object.put("stopped", BackgroundJobController.getInstance(context).isInterrupted(backgroundJobId));
+ object.put("jobId", jobController.getJob().getId());
+ object.put("backgroundJobId", backgroundJobId);
+ object.put("operationType", "CSV2HDFS");
+ object.put("status", BackgroundJobController.getInstance(context).state(backgroundJobId).toString());
+
+ return Response.ok(object).build();
+ } catch (WebApplicationException ex) {
+ throw ex;
+ } catch (ItemNotFound itemNotFound) {
+ throw new NotFoundFormattedException(itemNotFound.getMessage(), itemNotFound);
+ } catch (Exception ex) {
+ throw new ServiceFormattedException(ex.getMessage(), ex);
+ }
+ }
+
+
+ @Path("{jobId}/status")
+ @GET
+ @Consumes(MediaType.APPLICATION_JSON)
+ @Produces(MediaType.APPLICATION_JSON)
+ public Response fetchJobStatus(@PathParam("jobId") String jobId) throws ItemNotFound, HiveClientException, NoOperationStatusSetException {
+ JobController jobController = getResourceManager().readController(jobId);
+ Job job = jobController.getJob();
+ String jobStatus = job.getStatus();
+
+
+ LOG.info("jobStatus : {} for jobId : {}",jobStatus, jobId);
+
+ JSONObject jsonObject = new JSONObject();
+ jsonObject.put("jobStatus", jobStatus);
+ jsonObject.put("jobId", jobId);
+
+ return Response.ok(jsonObject).build();
+ }
+
+ /**
+ * Get next results page
+ */
+ @GET
+ @Path("{jobId}/results")
+ @Produces(MediaType.APPLICATION_JSON)
+ public Response getResults(@PathParam("jobId") final String jobId,
+ @QueryParam("first") final String fromBeginning,
+ @QueryParam("count") Integer count,
+ @QueryParam("searchId") String searchId,
+ @QueryParam("format") String format,
+ @QueryParam("columns") final String requestedColumns) {
+ try {
+
+ final String username = context.getUsername();
+
+ ConnectionSystem system = ConnectionSystem.getInstance();
+ final AsyncJobRunner asyncJobRunner = new AsyncJobRunnerImpl(context, system.getOperationController(context), system.getActorSystem());
+
+ return ResultsPaginationController.getInstance(context)
+ .request(jobId, searchId, true, fromBeginning, count, format,requestedColumns,
+ new Callable<Cursor< Row, ColumnDescription >>() {
+ @Override
+ public Cursor call() throws Exception {
+ Optional<NonPersistentCursor> cursor;
+ if(fromBeginning != null && fromBeginning.equals("true")){
+ cursor = asyncJobRunner.resetAndGetCursor(jobId, username);
+ }
+ else {
+ cursor = asyncJobRunner.getCursor(jobId, username);
+ }
+ if(cursor.isPresent())
+ return cursor.get();
+ else
+ return new EmptyCursor();
+ }
+ }).build();
+
+ } catch (WebApplicationException ex) {
+ throw ex;
+ } catch (Exception ex) {
+ throw new ServiceFormattedException(ex.getMessage(), ex);
+ }
+ }
+
+ /**
+ * Renew expiration time for results
+ */
+ @GET
+ @Path("{jobId}/results/keepAlive")
+ public Response keepAliveResults(@PathParam("jobId") String jobId,
+ @QueryParam("first") String fromBeginning,
+ @QueryParam("count") Integer count) {
+ try {
+ if (!ResultsPaginationController.getInstance(context).keepAlive(jobId, ResultsPaginationController.DEFAULT_SEARCH_ID)) {
+ throw new NotFoundFormattedException("Results already expired", null);
+ }
+ return Response.ok().build();
+ } catch (WebApplicationException ex) {
+ throw ex;
+ } catch (Exception ex) {
+ throw new ServiceFormattedException(ex.getMessage(), ex);
+ }
+ }
+
+ /**
+ * Get progress info
+ */
+ @GET
+ @Path("{jobId}/progress")
+ @Produces(MediaType.APPLICATION_JSON)
+ public Response getProgress(@PathParam("jobId") String jobId) {
+ try {
+ final JobController jobController = getResourceManager().readController(jobId);
+
+ ProgressRetriever.Progress progress = new ProgressRetriever(jobController.getJob(), getSharedObjectsFactory()).
+ getProgress();
+
+ return Response.ok(progress).build();
+ } catch (WebApplicationException ex) {
+ throw ex;
+ } catch (ItemNotFound itemNotFound) {
+ throw new NotFoundFormattedException(itemNotFound.getMessage(), itemNotFound);
+ } catch (Exception ex) {
+ throw new ServiceFormattedException(ex.getMessage(), ex);
+ }
+ }
+
+ /**
+ * Delete single item
+ */
+ @DELETE
+ @Path("{id}")
+ public Response delete(@PathParam("id") String id,
+ @QueryParam("remove") final String remove) {
+ try {
+ JobController jobController;
+ try {
+ jobController = getResourceManager().readController(id);
+ } catch (ItemNotFound itemNotFound) {
+ throw new NotFoundFormattedException(itemNotFound.getMessage(), itemNotFound);
+ }
+ jobController.cancel();
+ if (remove != null && remove.compareTo("true") == 0) {
+ getResourceManager().delete(id);
+ }
+ return Response.status(204).build();
+ } catch (WebApplicationException ex) {
+ throw ex;
+ } catch (ItemNotFound itemNotFound) {
+ throw new NotFoundFormattedException(itemNotFound.getMessage(), itemNotFound);
+ } catch (Exception ex) {
+ throw new ServiceFormattedException(ex.getMessage(), ex);
+ }
+ }
+
+ /**
+ * Get all Jobs
+ */
+ @GET
+ @Produces(MediaType.APPLICATION_JSON)
+ public Response getList() {
+ try {
+ LOG.debug("Getting all job");
+ List<Job> allJobs = getAggregator().readAll(context.getUsername());
+ for(Job job : allJobs) {
+ job.setSessionTag(null);
+ }
+
+ JSONObject object = new JSONObject();
+ object.put("jobs", allJobs);
+ return Response.ok(object).build();
+ } catch (WebApplicationException ex) {
+ throw ex;
+ } catch (Exception ex) {
+ throw new ServiceFormattedException(ex.getMessage(), ex);
+ }
+ }
+
+ /**
+ * Create job
+ */
+ @POST
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response create(JobRequest request, @Context HttpServletResponse response,
+ @Context UriInfo ui) {
+ try {
+ Map jobInfo = PropertyUtils.describe(request.job);
+ Job job = new JobImpl(jobInfo);
+
+
+ getResourceManager().create(job);
+
+ JobController createdJobController = getResourceManager().readController(job.getId());
+ createdJobController.submit();
+ getResourceManager().saveIfModified(createdJobController);
+
+ response.setHeader("Location",
+ String.format("%s/%s", ui.getAbsolutePath().toString(), job.getId()));
+
+ JSONObject jobObject = jsonObjectFromJob(createdJobController);
+
+ return Response.ok(jobObject).status(201).build();
+ } catch (WebApplicationException ex) {
+ throw ex;
+ } catch (ItemNotFound itemNotFound) {
+ throw new NotFoundFormattedException(itemNotFound.getMessage(), itemNotFound);
+ } catch (Throwable ex) {
+ throw new ServiceFormattedException(ex.getMessage(), ex);
+ }
+ }
+
+ /**
+ * Set password and connect to Hive
+ */
+ @POST
+ @Path("auth")
+ @Consumes(MediaType.APPLICATION_JSON)
+ public Response setupPassword(AuthRequest request) {
+ try {
+ /*HiveAuthCredentials authCredentials = new HiveAuthCredentials();
+ authCredentials.setPassword(request.password);
+ new UserLocalHiveAuthCredentials().set(authCredentials, context);*/
+
+ //connectionLocal.remove(context); // force reconnect on next get
+ //connectionLocal.get(context);
+ return Response.ok().status(200).build();
+ } catch (WebApplicationException ex) {
+ throw ex;
+ } catch (Exception ex) {
+ throw new ServiceFormattedException(ex.getMessage(), ex);
+ }
+ }
+
+ /**
+ * Remove connection credentials
+ */
+ @DELETE
+ @Path("auth")
+ public Response removePassword() {
+ try {
+ //new UserLocalHiveAuthCredentials().remove(context);
+ //connectionLocal.remove(context); // force reconnect on next get
+ return Response.ok().status(200).build();
+ } catch (WebApplicationException ex) {
+ throw ex;
+ } catch (Exception ex) {
+ throw new ServiceFormattedException(ex.getMessage(), ex);
+ }
+ }
+
+
+ /**
+ * Invalidate session
+ */
+ @DELETE
+ @Path("sessions/{sessionTag}")
+ public Response invalidateSession(@PathParam("sessionTag") String sessionTag) {
+ try {
+ //Connection connection = connectionLocal.get(context);
+ //connection.invalidateSessionByTag(sessionTag);
+ return Response.ok().build();
+ } catch (WebApplicationException ex) {
+ throw ex;
+ } catch (Exception ex) {
+ throw new ServiceFormattedException(ex.getMessage(), ex);
+ }
+ }
+
+ /**
+ * Session status
+ */
+ @GET
+ @Path("sessions/{sessionTag}")
+ @Produces(MediaType.APPLICATION_JSON)
+ public Response sessionStatus(@PathParam("sessionTag") String sessionTag) {
+ try {
+ //Connection connection = connectionLocal.get(context);
+
+ JSONObject session = new JSONObject();
+ session.put("sessionTag", sessionTag);
+ try {
+ //connection.getSessionByTag(sessionTag);
+ session.put("actual", true);
+ } catch (Exception /*HiveClientException*/ ex) {
+ session.put("actual", false);
+ }
+
+ //TODO: New implementation
+
+ JSONObject status = new JSONObject();
+ status.put("session", session);
+ return Response.ok(status).build();
+ } catch (WebApplicationException ex) {
+ throw ex;
+ } catch (Exception ex) {
+ throw new ServiceFormattedException(ex.getMessage(), ex);
+ }
+ }
+
+ /**
+ * Wrapper object for json mapping
+ */
+ public static class JobRequest {
+ public JobImpl job;
+ }
+
+ /**
+ * Wrapper for authentication json mapping
+ */
+ public static class AuthRequest {
+ public String password;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/b88db3cc/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/ModifyNotificationDelegate.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/ModifyNotificationDelegate.java b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/ModifyNotificationDelegate.java
new file mode 100644
index 0000000..aad346b
--- /dev/null
+++ b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/ModifyNotificationDelegate.java
@@ -0,0 +1,23 @@
+/**
+ * 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.hive2.resources.jobs;
+
+public interface ModifyNotificationDelegate {
+ boolean onModification(Object object);
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/b88db3cc/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/ModifyNotificationInvocationHandler.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/ModifyNotificationInvocationHandler.java b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/ModifyNotificationInvocationHandler.java
new file mode 100644
index 0000000..e4d6739
--- /dev/null
+++ b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/ModifyNotificationInvocationHandler.java
@@ -0,0 +1,40 @@
+/**
+ * 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.hive2.resources.jobs;
+
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.Method;
+
+public class ModifyNotificationInvocationHandler implements InvocationHandler {
+ private Object proxied;
+ private ModifyNotificationDelegate modifyDelegate;
+
+ public ModifyNotificationInvocationHandler(Object proxied, ModifyNotificationDelegate delegate) {
+ this.proxied = proxied;
+ this.modifyDelegate = delegate;
+ }
+
+ @Override
+ public Object invoke(Object o, Method method, Object[] args) throws Throwable {
+ if (method.getName().startsWith("set")) {
+ modifyDelegate.onModification(proxied);
+ }
+ return method.invoke(proxied, args);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/b88db3cc/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/NoOperationStatusSetException.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/NoOperationStatusSetException.java b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/NoOperationStatusSetException.java
new file mode 100644
index 0000000..1b92878
--- /dev/null
+++ b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/NoOperationStatusSetException.java
@@ -0,0 +1,23 @@
+/**
+ * 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.hive2.resources.jobs;
+
+
+public class NoOperationStatusSetException extends Exception {
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/b88db3cc/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/ProgressRetriever.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/ProgressRetriever.java b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/ProgressRetriever.java
new file mode 100644
index 0000000..8f9caa4
--- /dev/null
+++ b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/ProgressRetriever.java
@@ -0,0 +1,66 @@
+/**
+ * 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.hive2.resources.jobs;
+
+import org.apache.ambari.view.hive2.resources.jobs.atsJobs.TezVertexId;
+import org.apache.ambari.view.hive2.resources.jobs.rm.RMParser;
+import org.apache.ambari.view.hive2.resources.jobs.viewJobs.Job;
+import org.apache.ambari.view.hive2.utils.ServiceFormattedException;
+import org.apache.ambari.view.hive2.utils.SharedObjectsFactory;
+
+import java.util.List;
+
+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/b88db3cc/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/ResultsPaginationController.java
----------------------------------------------------------------------
diff --git a/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/ResultsPaginationController.java b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/ResultsPaginationController.java
new file mode 100644
index 0000000..6131d03
--- /dev/null
+++ b/contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/resources/jobs/ResultsPaginationController.java
@@ -0,0 +1,284 @@
+/**
+ * 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.hive2.resources.jobs;
+
+
+import com.google.common.base.Function;
+import com.google.common.collect.FluentIterable;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+import org.apache.ambari.view.ViewContext;
+import org.apache.ambari.view.hive2.client.ColumnDescription;
+import org.apache.ambari.view.hive2.client.Cursor;
+import org.apache.ambari.view.hive2.client.HiveClientException;
+import org.apache.ambari.view.hive2.client.Row;
+import org.apache.ambari.view.hive2.utils.BadRequestFormattedException;
+import org.apache.ambari.view.hive2.utils.HiveClientFormattedException;
+import org.apache.ambari.view.hive2.utils.ServiceFormattedException;
+import org.apache.commons.collections4.map.PassiveExpiringMap;
+import org.apache.hadoop.hbase.util.Strings;
+
+import javax.ws.rs.core.Response;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+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<Row, ColumnDescription>> resultsCache;
+
+ public static class CustomTimeToLiveExpirationPolicy extends PassiveExpiringMap.ConstantTimeToLiveExpirationPolicy<String, Cursor<Row, ColumnDescription>> {
+ public CustomTimeToLiveExpirationPolicy(long timeToLiveMillis) {
+ super(timeToLiveMillis);
+ }
+
+ @Override
+ public long expirationTime(String key, Cursor<Row, ColumnDescription> value) {
+ if (key.startsWith("$")) {
+ return -1; //never expire
+ }
+ return super.expirationTime(key, value);
+ }
+ }
+
+ private Map<String, Cursor<Row, ColumnDescription>> getResultsCache() {
+ if (resultsCache == null) {
+ PassiveExpiringMap<String, Cursor<Row, ColumnDescription>> resultsCacheExpiringMap =
+ new PassiveExpiringMap<>(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);
+ cursor.keepAlive();
+ return true;
+ }
+
+ private Cursor<Row, ColumnDescription> getResultsSet(String key, Callable<Cursor<Row, ColumnDescription>> makeResultsSet) {
+ if (!getResultsCache().containsKey(key)) {
+ Cursor resultSet;
+ try {
+ resultSet = makeResultsSet.call();
+ if (resultSet.isResettable()) {
+ resultSet.reset();
+ }
+ } 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, String requestedColumns, Callable<Cursor<Row, ColumnDescription>> 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<Row, ColumnDescription> resultSet = getResultsSet(key, makeResultsSet);
+
+ if (count == null)
+ count = DEFAULT_FETCH_COUNT;
+
+ List<ColumnDescription> allschema = resultSet.getDescriptions();
+ List<Row> allRowEntries = FluentIterable.from(resultSet)
+ .limit(count).toList();
+
+ List<ColumnDescription> schema = allschema;
+
+ final Set<Integer> selectedColumns = getRequestedColumns(requestedColumns);
+ if (!selectedColumns.isEmpty()) {
+ schema = filter(allschema, selectedColumns);
+ }
+
+ List<Object[]> rows = FluentIterable.from(allRowEntries)
+ .transform(new Function<Row, Object[]>() {
+ @Override
+ public Object[] apply(Row input) {
+ if(!selectedColumns.isEmpty()) {
+ return filter(Lists.newArrayList(input.getRow()), selectedColumns).toArray();
+ } else {
+ return input.getRow();
+ }
+ }
+ }).toList();
+
+ int read = rows.size();
+ if(format != null && format.equalsIgnoreCase("d3")) {
+ List<Map<String,Object>> results = new ArrayList<>();
+ for(int i=0; i<rows.size(); i++) {
+ Object[] row = rows.get(i);
+ Map<String, Object> keyValue = new HashMap<>(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);
+ }
+ }
+
+ private <T> List<T> filter(List<T> list, Set<Integer> selectedColumns) {
+ List<T> filtered = Lists.newArrayList();
+ for(int i: selectedColumns) {
+ filtered.add(list.get(i));
+ }
+
+ return filtered;
+ }
+
+ private Set<Integer> getRequestedColumns(String requestedColumns) {
+ if(Strings.isEmpty(requestedColumns)) {
+ return new HashSet<>();
+ }
+ Set<Integer> selectedColumns = Sets.newHashSet();
+ for (String columnRequested : requestedColumns.split(",")) {
+ try {
+ selectedColumns.add(Integer.parseInt(columnRequested));
+ } catch (NumberFormatException ex) {
+ throw new BadRequestFormattedException("Columns param should be comma-separated integers", ex);
+ }
+ }
+ return selectedColumns;
+ }
+
+ private static class ResultsResponse {
+ private List<ColumnDescription> schema;
+ private List<String[]> rows;
+ private int readCount;
+ private boolean hasNext;
+ private long offset;
+ private boolean hasResults;
+
+ public void setSchema(List<ColumnDescription> schema) {
+ this.schema = schema;
+ }
+
+ public List<ColumnDescription> getSchema() {
+ return schema;
+ }
+
+ public void setRows(List<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 List<String[]> getRows() {
+ return rows;
+ }
+
+ public void setReadCount(int readCount) {
+ this.readCount = 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;
+ }
+ }
+}