Repository: sqoop Updated Branches: refs/heads/sqoop2 c820aaf87 -> 9151d305a
SQOOP-1839: Sqoop2: Mapreduce submission record should propogate failure info as well (Veena Basavaraj via Abraham Elmahrek) Project: http://git-wip-us.apache.org/repos/asf/sqoop/repo Commit: http://git-wip-us.apache.org/repos/asf/sqoop/commit/9151d305 Tree: http://git-wip-us.apache.org/repos/asf/sqoop/tree/9151d305 Diff: http://git-wip-us.apache.org/repos/asf/sqoop/diff/9151d305 Branch: refs/heads/sqoop2 Commit: 9151d305affa0cd7092e3343ebcfbff9b274510d Parents: c820aaf Author: Abraham Elmahrek <[email protected]> Authored: Thu Dec 11 01:43:34 2014 -0600 Committer: Abraham Elmahrek <[email protected]> Committed: Thu Dec 11 01:43:34 2014 -0600 ---------------------------------------------------------------------- .../org/apache/sqoop/json/SubmissionBean.java | 26 +++---- .../org/apache/sqoop/model/MSubmission.java | 76 ++++++-------------- .../org/apache/sqoop/model/SubmissionError.java | 75 +++++++++++++++++++ .../apache/sqoop/json/TestSubmissionBean.java | 48 +++++++------ .../org/apache/sqoop/driver/JobManager.java | 23 +++--- .../org/apache/sqoop/driver/JobRequest.java | 12 ++-- .../apache/sqoop/driver/SubmissionEngine.java | 31 +++++--- docs/src/site/sphinx/RESTAPI.rst | 10 +-- .../common/CommonRepositoryHandler.java | 20 +++--- .../derby/TestSubmissionHandling.java | 12 ++-- .../sqoop/shell/ShowSubmissionFunction.java | 2 +- .../sqoop/shell/utils/SubmissionDisplayer.java | 10 +-- .../mapreduce/MapreduceSubmissionEngine.java | 70 +++++++++++++----- .../sqoop/test/testcases/ConnectorTestCase.java | 7 +- 14 files changed, 259 insertions(+), 163 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/sqoop/blob/9151d305/common/src/main/java/org/apache/sqoop/json/SubmissionBean.java ---------------------------------------------------------------------- diff --git a/common/src/main/java/org/apache/sqoop/json/SubmissionBean.java b/common/src/main/java/org/apache/sqoop/json/SubmissionBean.java index e926f02..66db0d4 100644 --- a/common/src/main/java/org/apache/sqoop/json/SubmissionBean.java +++ b/common/src/main/java/org/apache/sqoop/json/SubmissionBean.java @@ -48,8 +48,8 @@ public class SubmissionBean implements JsonBean { private static final String STATUS = "status"; private static final String EXTERNAL_ID = "external-id"; private static final String EXTERNAL_LINK = "external-link"; - private static final String EXCEPTION = "exception"; - private static final String EXCEPTION_TRACE = "exception-trace"; + private static final String ERROR_SUMMARY = "error-summary"; + private static final String ERROR_DETAILS = "error-details"; private static final String PROGRESS = "progress"; private static final String COUNTERS = "counters"; private static final String FROM_SCHEMA = "from-schema"; @@ -114,17 +114,17 @@ public class SubmissionBean implements JsonBean { if (submission.getLastUpdateDate() != null) { object.put(LAST_UPDATE_DATE, submission.getLastUpdateDate().getTime()); } - if (submission.getExternalId() != null) { - object.put(EXTERNAL_ID, submission.getExternalId()); + if (submission.getExternalJobId() != null) { + object.put(EXTERNAL_ID, submission.getExternalJobId()); } if (submission.getExternalLink() != null) { object.put(EXTERNAL_LINK, submission.getExternalLink()); } - if (submission.getExceptionInfo() != null) { - object.put(EXCEPTION, submission.getExceptionInfo()); + if (submission.getError().getErrorSummary() != null) { + object.put(ERROR_SUMMARY, submission.getError().getErrorSummary()); } - if (submission.getExceptionStackTrace() != null) { - object.put(EXCEPTION_TRACE, submission.getExceptionStackTrace()); + if (submission.getError().getErrorDetails() != null) { + object.put(ERROR_DETAILS, submission.getError().getErrorDetails()); } if (submission.getCounters() != null) { object.put(COUNTERS, extractCounters(submission.getCounters())); @@ -186,16 +186,16 @@ public class SubmissionBean implements JsonBean { submission.setLastUpdateDate(new Date((Long) object.get(LAST_UPDATE_DATE))); } if (object.containsKey(EXTERNAL_ID)) { - submission.setExternalId((String) object.get(EXTERNAL_ID)); + submission.setExternalJobId((String) object.get(EXTERNAL_ID)); } if (object.containsKey(EXTERNAL_LINK)) { submission.setExternalLink((String) object.get(EXTERNAL_LINK)); } - if (object.containsKey(EXCEPTION)) { - submission.setExceptionInfo((String) object.get(EXCEPTION)); + if (object.containsKey(ERROR_SUMMARY)) { + submission.getError().setErrorSummary((String) object.get(ERROR_SUMMARY)); } - if (object.containsKey(EXCEPTION_TRACE)) { - submission.setExceptionStackTrace((String) object.get(EXCEPTION_TRACE)); + if (object.containsKey(ERROR_DETAILS)) { + submission.getError().setErrorDetails((String) object.get(ERROR_DETAILS)); } if (object.containsKey(COUNTERS)) { submission.setCounters(restoreCounters((JSONObject) object.get(COUNTERS))); http://git-wip-us.apache.org/repos/asf/sqoop/blob/9151d305/common/src/main/java/org/apache/sqoop/model/MSubmission.java ---------------------------------------------------------------------- diff --git a/common/src/main/java/org/apache/sqoop/model/MSubmission.java b/common/src/main/java/org/apache/sqoop/model/MSubmission.java index 2648712..ff50c99 100644 --- a/common/src/main/java/org/apache/sqoop/model/MSubmission.java +++ b/common/src/main/java/org/apache/sqoop/model/MSubmission.java @@ -21,8 +21,6 @@ import org.apache.sqoop.schema.Schema; import org.apache.sqoop.submission.SubmissionStatus; import org.apache.sqoop.submission.counter.Counters; -import java.io.PrintWriter; -import java.io.StringWriter; import java.util.Date; /** @@ -48,12 +46,13 @@ public class MSubmission extends MAccountableEntity { SubmissionStatus status; /** - * Any valid external id associated with this submission. + * Any valid external id associated with this job submission. + * For instance: in the case of MR execution engine, this will refer to the MR job Id * * This property is optional and might be NULL in case that the job has not * yet been submitted to the external system (JobTracker, ResourceManager, ...). */ - String externalId; + String externalJobId; /** * Progress in the job. @@ -83,18 +82,11 @@ public class MSubmission extends MAccountableEntity { String externalLink; /** - * Associated exception info with this job (if any). + * Associated error (exception or failure) with this job (if any). * * This property is optional. */ - String exceptionInfo; - - /** - * Associated exception stacktrace with this job (if any). - * - * This property is optional. - */ - String exceptionStackTrace; + SubmissionError error; /** * Schema for the FROM part of the job submission @@ -129,7 +121,7 @@ public class MSubmission extends MAccountableEntity { public MSubmission(long jobId, Date creationDate, SubmissionStatus status, String externalId) { this(jobId, creationDate, status); - this.externalId = externalId; + this.externalJobId = externalId; } public MSubmission(long jobId, Date creationDate, SubmissionStatus status, @@ -155,12 +147,12 @@ public class MSubmission extends MAccountableEntity { return status; } - public void setExternalId(String externalId) { - this.externalId = externalId; + public void setExternalJobId(String externalJobId) { + this.externalJobId = externalJobId; } - public String getExternalId() { - return externalId; + public String getExternalJobId() { + return externalJobId; } public void setProgress(double progress) { @@ -187,31 +179,15 @@ public class MSubmission extends MAccountableEntity { return externalLink; } - public void setExceptionInfo(String exceptionInfo) { - this.exceptionInfo = exceptionInfo; - } - - public String getExceptionInfo() { - return exceptionInfo; + public void setError(SubmissionError error) { + this.error = error; } - public void setExceptionStackTrace(String stackTrace) { - this.exceptionStackTrace = stackTrace; - } - - public String getExceptionStackTrace() { - return exceptionStackTrace; - } - - public void setException(Throwable e) { - // Exception info - this.setExceptionInfo(e.toString()); - - // Exception stack trace - StringWriter writer = new StringWriter(); - e.printStackTrace(new PrintWriter(writer)); - writer.flush(); - this.setExceptionStackTrace(writer.toString()); + public SubmissionError getError() { + if(this.error == null) { + this.error = new SubmissionError(); + } + return this.error; } public Schema getFromSchema() { @@ -232,21 +208,11 @@ public class MSubmission extends MAccountableEntity { @Override public String toString() { - return "MSubmission{" + - "jobId=" + jobId + - ", creationDate=" + getCreationDate() + - ", lastUpdateDate=" + getLastUpdateDate() + - ", status=" + status + - ", externalId='" + externalId + '\'' + - ", progress=" + progress + - ", counters=" + counters + - ", externalLink='" + externalLink + '\'' + - ", exceptionInfo='" + exceptionInfo + '\'' + - ", exceptionStackTrace='" + exceptionStackTrace + '\'' + - ", fromSchema='" + fromSchema + '\'' + - ", toSchema='" + toSchema + '\'' + - '}'; + return "MSubmission [jobId=" + jobId + ", status=" + status + ", externalId=" + externalJobId + + ", progress=" + progress + ", counters=" + counters + ", externalLink=" + externalLink + + ", error=" + error + ", fromSchema=" + fromSchema + ", toSchema=" + toSchema + "]"; } public static MSubmission UNKNOWN = new MSubmission(); + } http://git-wip-us.apache.org/repos/asf/sqoop/blob/9151d305/common/src/main/java/org/apache/sqoop/model/SubmissionError.java ---------------------------------------------------------------------- diff --git a/common/src/main/java/org/apache/sqoop/model/SubmissionError.java b/common/src/main/java/org/apache/sqoop/model/SubmissionError.java new file mode 100644 index 0000000..698877d --- /dev/null +++ b/common/src/main/java/org/apache/sqoop/model/SubmissionError.java @@ -0,0 +1,75 @@ +/** + * 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.sqoop.model; + +/** + *Represents the job submission error + */ +public class SubmissionError { + + /** + * Associated error (exception or failure) summary with this job (if any). + * + * This property is optional. + */ + String errorSummary; + + /** + * Associated error details, possibly the stacktrace with this job (if any). + * + * This property is optional. + */ + String errorDetails; + + /** + * @return the errorSummary + */ + public String getErrorSummary() { + return errorSummary; + } + + /** + * @param errorSummary + * the errorSummary to set + */ + public void setErrorSummary(String errorSummary) { + this.errorSummary = errorSummary; + } + + /** + * @return the errorDetails + */ + public String getErrorDetails() { + return errorDetails; + } + + /** + * @param errorDetails + * the errorDetails to set + */ + public void setErrorDetails(String errorDetails) { + this.errorDetails = errorDetails; + } + + @Override + public String toString() { + return "JobSubmissionError [errorSummary=" + errorSummary + ", errorDetails=" + errorDetails + + "]"; + } + +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/sqoop/blob/9151d305/common/src/test/java/org/apache/sqoop/json/TestSubmissionBean.java ---------------------------------------------------------------------- diff --git a/common/src/test/java/org/apache/sqoop/json/TestSubmissionBean.java b/common/src/test/java/org/apache/sqoop/json/TestSubmissionBean.java index d07eda9..24c0b34 100644 --- a/common/src/test/java/org/apache/sqoop/json/TestSubmissionBean.java +++ b/common/src/test/java/org/apache/sqoop/json/TestSubmissionBean.java @@ -17,6 +17,7 @@ */ package org.apache.sqoop.json; +import org.apache.sqoop.model.SubmissionError; import org.apache.sqoop.model.MSubmission; import org.apache.sqoop.schema.Schema; import org.apache.sqoop.schema.type.Decimal; @@ -199,24 +200,24 @@ public class TestSubmissionBean { @Test public void testTransferExternalId() { MSubmission source = new MSubmission(); - source.setExternalId("Job-x"); + source.setExternalJobId("Job-x"); MSubmission target = transfer(source); - assertEquals("Job-x", target.getExternalId()); + assertEquals("Job-x", target.getExternalJobId()); List<MSubmission> sources = new ArrayList<MSubmission>(); MSubmission sourcex = new MSubmission(); - sourcex.setExternalId("Job-y"); + sourcex.setExternalJobId("Job-y"); sources.add(sourcex); MSubmission sourcey = new MSubmission(); - sourcey.setExternalId("Job-z"); + sourcey.setExternalJobId("Job-z"); sources.add(sourcey); List<MSubmission> targets = transfer(sources); assertNotNull(targets.get(0)); - assertEquals("Job-y", targets.get(0).getExternalId()); + assertEquals("Job-y", targets.get(0).getExternalJobId()); assertNotNull(targets.get(1)); - assertEquals("Job-z", targets.get(1).getExternalId()); + assertEquals("Job-z", targets.get(1).getExternalJobId()); } @Test @@ -243,49 +244,54 @@ public class TestSubmissionBean { } @Test - public void testTransferException() { + public void testTransferErrorSummary() { + SubmissionError error = new SubmissionError(); MSubmission source = new MSubmission(); - source.setExceptionInfo("EndOfTheWorldException"); - + error.setErrorSummary("EndOfTheWorldException"); + source.setError(error); MSubmission target = transfer(source); - assertEquals("EndOfTheWorldException", target.getExceptionInfo()); + assertEquals("EndOfTheWorldException", target.getError().getErrorSummary()); List<MSubmission> sources = new ArrayList<MSubmission>(); MSubmission sourcex = new MSubmission(); - sourcex.setExceptionInfo("TheNewEraException"); + SubmissionError errorx= new SubmissionError(); + errorx.setErrorSummary("TheNewEraException"); + sourcex.setError(errorx); sources.add(sourcex); MSubmission sourcey = new MSubmission(); - sourcey.setExceptionInfo("EndOfTheWorldAgainException"); + SubmissionError errory= new SubmissionError(); + errory.setErrorSummary("EndOfTheWorldAgainException"); + sourcey.setError(errory); sources.add(sourcey); List<MSubmission> targets = transfer(sources); assertNotNull(targets.get(0)); - assertEquals("TheNewEraException", targets.get(0).getExceptionInfo()); + assertEquals("TheNewEraException", targets.get(0).getError().getErrorSummary()); assertNotNull(targets.get(1)); - assertEquals("EndOfTheWorldAgainException", targets.get(1).getExceptionInfo()); + assertEquals("EndOfTheWorldAgainException", targets.get(1).getError().getErrorSummary()); } @Test - public void testTransferExceptionTrace() { + public void testTransferErrorDetails() { MSubmission source = new MSubmission(); - source.setExceptionStackTrace("void.java(3): line infinity"); + source.getError().setErrorDetails("void.java(3): line infinity"); MSubmission target = transfer(source); - assertEquals("void.java(3): line infinity", target.getExceptionStackTrace()); + assertEquals("void.java(3): line infinity", target.getError().getErrorDetails()); List<MSubmission> sources = new ArrayList<MSubmission>(); MSubmission sourcex = new MSubmission(); - sourcex.setExceptionStackTrace("void.java(4): segment fault in Java"); + sourcex.getError().setErrorDetails("void.java(4): segment fault in Java"); sources.add(sourcex); MSubmission sourcey = new MSubmission(); - sourcey.setExceptionStackTrace("void.java(5): core dumps in Java"); + sourcey.getError().setErrorDetails("void.java(5): core dumps in Java"); sources.add(sourcey); List<MSubmission> targets = transfer(sources); assertNotNull(targets.get(0)); - assertEquals("void.java(4): segment fault in Java", targets.get(0).getExceptionStackTrace()); + assertEquals("void.java(4): segment fault in Java", targets.get(0).getError().getErrorDetails()); assertNotNull(targets.get(1)); - assertEquals("void.java(5): core dumps in Java", targets.get(1).getExceptionStackTrace()); + assertEquals("void.java(5): core dumps in Java", targets.get(1).getError().getErrorDetails()); } @Test http://git-wip-us.apache.org/repos/asf/sqoop/blob/9151d305/core/src/main/java/org/apache/sqoop/driver/JobManager.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/sqoop/driver/JobManager.java b/core/src/main/java/org/apache/sqoop/driver/JobManager.java index f4f5561..2709512 100644 --- a/core/src/main/java/org/apache/sqoop/driver/JobManager.java +++ b/core/src/main/java/org/apache/sqoop/driver/JobManager.java @@ -37,6 +37,7 @@ import org.apache.sqoop.job.etl.Initializer; import org.apache.sqoop.job.etl.InitializerContext; import org.apache.sqoop.job.etl.Transferable; import org.apache.sqoop.model.ConfigUtils; +import org.apache.sqoop.model.SubmissionError; import org.apache.sqoop.model.MJob; import org.apache.sqoop.model.MLink; import org.apache.sqoop.model.MSubmission; @@ -373,8 +374,8 @@ public class JobManager implements Reconfigurable { initializeConnector(jobRequest, Direction.FROM); initializeConnector(jobRequest, Direction.TO); - jobRequest.getSummary().setFromSchema(getSchemaForConnector(jobRequest, Direction.FROM)); - jobRequest.getSummary().setToSchema(getSchemaForConnector(jobRequest, Direction.TO)); + jobRequest.getJobSubmission().setFromSchema(getSchemaForConnector(jobRequest, Direction.FROM)); + jobRequest.getJobSubmission().setToSchema(getSchemaForConnector(jobRequest, Direction.TO)); LOG.debug("Using entities: " + jobRequest.getFrom() + ", " + jobRequest.getTo()); return jobRequest; @@ -531,10 +532,10 @@ public class JobManager implements Reconfigurable { } DestroyerContext fromDestroyerContext = new DestroyerContext( - request.getConnectorContext(Direction.FROM), false, request.getSummary() + request.getConnectorContext(Direction.FROM), false, request.getJobSubmission() .getFromSchema()); DestroyerContext toDestroyerContext = new DestroyerContext( - request.getConnectorContext(Direction.TO), false, request.getSummary() + request.getConnectorContext(Direction.TO), false, request.getJobSubmission() .getToSchema()); fromDestroyer.destroy(fromDestroyerContext, request.getConnectorLinkConfig(Direction.FROM), @@ -552,7 +553,7 @@ public class JobManager implements Reconfigurable { throw new SqoopException(DriverError.DRIVER_0003, "Job with id " + jobId + " is not running hence cannot stop"); } - submissionEngine.stop(mSubmission.getExternalId()); + submissionEngine.stop(mSubmission.getExternalJobId()); mSubmission.setLastUpdateUser(ctx.getUsername()); @@ -581,17 +582,19 @@ public class JobManager implements Reconfigurable { private void update(MSubmission submission) { double progress = -1; Counters counters = null; - String externalId = submission.getExternalId(); - SubmissionStatus newStatus = submissionEngine.status(externalId); - String externalLink = submissionEngine.externalLink(externalId); + String externalJobId = submission.getExternalJobId(); + SubmissionStatus newStatus = submissionEngine.status(externalJobId); + SubmissionError error = submissionEngine.error(externalJobId); + String externalLink = submissionEngine.externalLink(externalJobId); if (newStatus.isRunning()) { - progress = submissionEngine.progress(externalId); + progress = submissionEngine.progress(externalJobId); } else { - counters = submissionEngine.counters(externalId); + counters = submissionEngine.counters(externalJobId); } submission.setStatus(newStatus); + submission.setError(error); submission.setProgress(progress); submission.setCounters(counters); submission.setExternalLink(externalLink); http://git-wip-us.apache.org/repos/asf/sqoop/blob/9151d305/core/src/main/java/org/apache/sqoop/driver/JobRequest.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/sqoop/driver/JobRequest.java b/core/src/main/java/org/apache/sqoop/driver/JobRequest.java index eed79a5..8c1cc95 100644 --- a/core/src/main/java/org/apache/sqoop/driver/JobRequest.java +++ b/core/src/main/java/org/apache/sqoop/driver/JobRequest.java @@ -38,9 +38,9 @@ import java.util.List; public class JobRequest { /** - * Submission summary + * Job Submission */ - MSubmission summary; + MSubmission jobSubmission; /** * Original job name @@ -129,12 +129,12 @@ public class JobRequest { this.driverConfig = null; } - public MSubmission getSummary() { - return summary; + public MSubmission getJobSubmission() { + return jobSubmission; } - public void setSummary(MSubmission summary) { - this.summary = summary; + public void setSummary(MSubmission submission) { + this.jobSubmission = submission; } public String getJobName() { http://git-wip-us.apache.org/repos/asf/sqoop/blob/9151d305/core/src/main/java/org/apache/sqoop/driver/SubmissionEngine.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/sqoop/driver/SubmissionEngine.java b/core/src/main/java/org/apache/sqoop/driver/SubmissionEngine.java index 3a32e9f..62e0d8f 100644 --- a/core/src/main/java/org/apache/sqoop/driver/SubmissionEngine.java +++ b/core/src/main/java/org/apache/sqoop/driver/SubmissionEngine.java @@ -18,6 +18,7 @@ package org.apache.sqoop.driver; import org.apache.sqoop.common.MapContext; +import org.apache.sqoop.model.SubmissionError; import org.apache.sqoop.submission.counter.Counters; import org.apache.sqoop.submission.SubmissionStatus; @@ -58,22 +59,30 @@ public abstract class SubmissionEngine { * * @return Return true if we were able to submit job to remote cluster. */ - public abstract boolean submit(JobRequest submission); + public abstract boolean submit(JobRequest jobRequest); /** * Hard stop for given submission. * - * @param submissionId Submission internal id. + * @param externalJobId Submission external job id. */ - public abstract void stop(String submissionId); + public abstract void stop(String externalJobId); /** * Return status of given submission. * - * @param submissionId Submission internal id. + * @param externalJobId Submission external job id. * @return Current submission status. */ - public abstract SubmissionStatus status(String submissionId); + public abstract SubmissionStatus status(String externalJobId); + + /** + * Return failure info if the job status is FAILED + * + * @param submissionId Submission internal id. + * @return Current failure info + */ + public abstract SubmissionError error(String externalJobId); /** * Return submission progress. @@ -82,10 +91,10 @@ public abstract class SubmissionEngine { * has gone or -1 in case that this submission engine do not supports * progress reporting. * - * @param submissionId Submission internal id. + * @param externalJobId Submission external job id. * @return {-1} union <0, 1> */ - public double progress(String submissionId) { + public double progress(String externalJobId) { return -1; } @@ -95,21 +104,21 @@ public abstract class SubmissionEngine { * Sqoop will call counters only for submission in state SUCCEEDED, * it's consider exceptional state to call this method for other states. * - * @param submissionId Submission internal id. + * @param externalJobId Submission external job id. * @return Submission statistics */ - public Counters counters(String submissionId) { + public Counters counters(String externalJobId) { return null; } /** * Return link to external web page with given submission. * - * @param submissionId Submission internal id. + * @param externalJobId Submission external job id. * @return Null in case that external page is not supported or available or * HTTP link to given submission. */ - public String externalLink(String submissionId) { + public String externalLink(String externalJobId) { return null; } } http://git-wip-us.apache.org/repos/asf/sqoop/blob/9151d305/docs/src/site/sphinx/RESTAPI.rst ---------------------------------------------------------------------- diff --git a/docs/src/site/sphinx/RESTAPI.rst b/docs/src/site/sphinx/RESTAPI.rst index c0c236e..3d0c7c4 100644 --- a/docs/src/site/sphinx/RESTAPI.rst +++ b/docs/src/site/sphinx/RESTAPI.rst @@ -41,7 +41,7 @@ For instance a connector that represents a relational data source such as MySQL Each connector can support Reading from a data source and/or writing/to a data source it represents. Reading from and writing to a data source are represented by From and To respectively. Specific configurations are required to peform the job of reading from or writing to the data source. These are grouped in the ``FromJobConfiguration`` and ``ToJobConfiguration`` objects of the connector. -For instace, a connector that represents a relational data source such as MySQL will expose the table name to read from or the SQL query to use while reading data as a FromJobConfiguration. Similarly a connector that represents a data source such as HDFS, will expose the output directory to write to as a ToJobConfiguration. +For instance, a connector that represents a relational data source such as MySQL will expose the table name to read from or the SQL query to use while reading data as a FromJobConfiguration. Similarly a connector that represents a data source such as HDFS, will expose the output directory to write to as a ToJobConfiguration. Objects @@ -1268,7 +1268,7 @@ Start a job with name ``[jname]`` or with id ``[jid]`` to trigger the job execut "progress": -1, "last-update-date": 1415312390570, "status": "FAILURE_ON_SUBMIT", - "exception": "org.apache.sqoop.common.SqoopException: GENERIC_HDFS_CONNECTOR_0000:Error occurs during partitioner run", + "error-summary": "org.apache.sqoop.common.SqoopException: GENERIC_HDFS_CONNECTOR_0000:Error occurs during partitioner run", "job": 1, "creation-date": 1415312390570, "to-schema": { @@ -1296,7 +1296,7 @@ Start a job with name ``[jname]`` or with id ``[jid]`` to trigger the job execut "columns": [ ] }, - "exception-trace": "org.apache.sqoop.common.SqoopException: GENERIC_HDFS_CONNECTOR_00" + "error-details": "org.apache.sqoop.common.SqoopException: GENERIC_HDFS_CONNECTOR_00" } } @@ -1432,10 +1432,10 @@ Provide the name of the job in the url [jname] part. progress: -1, last-update-date: 1415312390570, status: "FAILURE_ON_SUBMIT", - exception: "org.apache.sqoop.common.SqoopException: GENERIC_HDFS_CONNECTOR_0000:Error occurs during partitioner run", + error-summary: "org.apache.sqoop.common.SqoopException: GENERIC_HDFS_CONNECTOR_0000:Error occurs during partitioner run", job: 1, creation-date: 1415312390570, - exception-trace: "org.apache.sqoop.common.SqoopException: GENERIC_HDFS_CONNECTOR_0000:Error occurs during partitioner...." + error-details: "org.apache.sqoop.common.SqoopException: GENERIC_HDFS_CONNECTOR_0000:Error occurs during partitioner...." } ] } \ No newline at end of file http://git-wip-us.apache.org/repos/asf/sqoop/blob/9151d305/repository/repository-common/src/main/java/org/apache/sqoop/repository/common/CommonRepositoryHandler.java ---------------------------------------------------------------------- diff --git a/repository/repository-common/src/main/java/org/apache/sqoop/repository/common/CommonRepositoryHandler.java b/repository/repository-common/src/main/java/org/apache/sqoop/repository/common/CommonRepositoryHandler.java index c278406..9fa2f9d 100644 --- a/repository/repository-common/src/main/java/org/apache/sqoop/repository/common/CommonRepositoryHandler.java +++ b/repository/repository-common/src/main/java/org/apache/sqoop/repository/common/CommonRepositoryHandler.java @@ -24,6 +24,7 @@ import org.apache.sqoop.common.DirectionError; import org.apache.sqoop.common.SqoopException; import org.apache.sqoop.common.SupportedDirections; import org.apache.sqoop.driver.Driver; +import org.apache.sqoop.model.SubmissionError; import org.apache.sqoop.model.MBooleanInput; import org.apache.sqoop.model.MConfig; import org.apache.sqoop.model.MConfigType; @@ -927,10 +928,10 @@ public abstract class CommonRepositoryHandler extends JdbcRepositoryHandler { stmt.setTimestamp(4, new Timestamp(submission.getCreationDate().getTime())); stmt.setString(5, submission.getLastUpdateUser()); stmt.setTimestamp(6, new Timestamp(submission.getLastUpdateDate().getTime())); - stmt.setString(7, submission.getExternalId()); + stmt.setString(7, submission.getExternalJobId()); stmt.setString(8, submission.getExternalLink()); - stmt.setString(9, submission.getExceptionInfo()); - stmt.setString(10, submission.getExceptionStackTrace()); + stmt.setString(9, submission.getError().getErrorSummary()); + stmt.setString(10, submission.getError().getErrorDetails()); result = stmt.executeUpdate(); if (result != 1) { @@ -999,8 +1000,8 @@ public abstract class CommonRepositoryHandler extends JdbcRepositoryHandler { stmt.setString(1, submission.getStatus().name()); stmt.setString(2, submission.getLastUpdateUser()); stmt.setTimestamp(3, new Timestamp(submission.getLastUpdateDate().getTime())); - stmt.setString(4, submission.getExceptionInfo()); - stmt.setString(5, submission.getExceptionStackTrace()); + stmt.setString(4, submission.getError().getErrorSummary()); + stmt.setString(5, submission.getError().getErrorDetails()); stmt.setLong(6, submission.getPersistenceId()); stmt.executeUpdate(); @@ -1441,11 +1442,12 @@ public abstract class CommonRepositoryHandler extends JdbcRepositoryHandler { submission.setCreationDate(rs.getTimestamp(5)); submission.setLastUpdateUser(rs.getString(6)); submission.setLastUpdateDate(rs.getTimestamp(7)); - submission.setExternalId(rs.getString(8)); + submission.setExternalJobId(rs.getString(8)); submission.setExternalLink(rs.getString(9)); - submission.setExceptionInfo(rs.getString(10)); - submission.setExceptionStackTrace(rs.getString(11)); - + SubmissionError error = new SubmissionError(); + error.setErrorSummary(rs.getString(10)); + error.setErrorDetails(rs.getString(11)); + submission.setError(error); Counters counters = loadCountersSubmission(rs.getLong(1), conn); submission.setCounters(counters); http://git-wip-us.apache.org/repos/asf/sqoop/blob/9151d305/repository/repository-derby/src/test/java/org/apache/sqoop/repository/derby/TestSubmissionHandling.java ---------------------------------------------------------------------- diff --git a/repository/repository-derby/src/test/java/org/apache/sqoop/repository/derby/TestSubmissionHandling.java b/repository/repository-derby/src/test/java/org/apache/sqoop/repository/derby/TestSubmissionHandling.java index e2e8073..dae5e73 100644 --- a/repository/repository-derby/src/test/java/org/apache/sqoop/repository/derby/TestSubmissionHandling.java +++ b/repository/repository-derby/src/test/java/org/apache/sqoop/repository/derby/TestSubmissionHandling.java @@ -111,10 +111,10 @@ public class TestSubmissionHandling extends DerbyTestCase { submission.setStatus(SubmissionStatus.RUNNING); submission.setCreationDate(creationDate); submission.setLastUpdateDate(updateDate); - submission.setExternalId("job-x"); + submission.setExternalJobId("job-x"); submission.setExternalLink("http://somewhere";); - submission.setExceptionInfo("RuntimeException"); - submission.setExceptionStackTrace("Yeah it happens"); + submission.getError().setErrorSummary("RuntimeException"); + submission.getError().setErrorDetails("Yeah it happens"); submission.setCounters(counters); handler.createSubmission(submission, getDerbyDatabaseConnection()); @@ -133,10 +133,10 @@ public class TestSubmissionHandling extends DerbyTestCase { assertEquals(SubmissionStatus.RUNNING, submission.getStatus()); assertEquals(creationDate, submission.getCreationDate()); assertEquals(updateDate, submission.getLastUpdateDate()); - assertEquals("job-x", submission.getExternalId()); + assertEquals("job-x", submission.getExternalJobId()); assertEquals("http://somewhere";, submission.getExternalLink()); - assertEquals("RuntimeException", submission.getExceptionInfo()); - assertEquals("Yeah it happens", submission.getExceptionStackTrace()); + assertEquals("RuntimeException", submission.getError().getErrorSummary()); + assertEquals("Yeah it happens", submission.getError().getErrorDetails()); CounterGroup group; Counter counter; http://git-wip-us.apache.org/repos/asf/sqoop/blob/9151d305/shell/src/main/java/org/apache/sqoop/shell/ShowSubmissionFunction.java ---------------------------------------------------------------------- diff --git a/shell/src/main/java/org/apache/sqoop/shell/ShowSubmissionFunction.java b/shell/src/main/java/org/apache/sqoop/shell/ShowSubmissionFunction.java index 4a0f476..af6de5b 100644 --- a/shell/src/main/java/org/apache/sqoop/shell/ShowSubmissionFunction.java +++ b/shell/src/main/java/org/apache/sqoop/shell/ShowSubmissionFunction.java @@ -84,7 +84,7 @@ public class ShowSubmissionFunction extends SqoopFunction { for (MSubmission submission : submissions) { jids.add(String.valueOf(submission.getJobId())); - eids.add(String.valueOf(submission.getExternalId())); + eids.add(String.valueOf(submission.getExternalJobId())); status.add(submission.getStatus().toString()); dates.add(submission.getLastUpdateDate().toString()); } http://git-wip-us.apache.org/repos/asf/sqoop/blob/9151d305/shell/src/main/java/org/apache/sqoop/shell/utils/SubmissionDisplayer.java ---------------------------------------------------------------------- diff --git a/shell/src/main/java/org/apache/sqoop/shell/utils/SubmissionDisplayer.java b/shell/src/main/java/org/apache/sqoop/shell/utils/SubmissionDisplayer.java index 0e2a38d..bb0268a 100644 --- a/shell/src/main/java/org/apache/sqoop/shell/utils/SubmissionDisplayer.java +++ b/shell/src/main/java/org/apache/sqoop/shell/utils/SubmissionDisplayer.java @@ -57,7 +57,7 @@ public final class SubmissionDisplayer { print(resourceString(Constants.RES_SUBMISSION_UPDATE_USER) + ": "); println(submission.getLastUpdateUser()); - String externalId = submission.getExternalId(); + String externalId = submission.getExternalJobId(); if(externalId != null) { print(resourceString(Constants.RES_SUBMISSION_EXTERNAL_ID)+": "); println(externalId); @@ -130,13 +130,13 @@ public final class SubmissionDisplayer { println(dateFormat.format(submission.getLastUpdateDate())+": "+submission.getStatus()); } // Exception handling - if (submission.getExceptionInfo() != null) { + if (submission.getError().getErrorSummary() != null) { print("@|red Exception: |@"); - println(submission.getExceptionInfo()); + println(submission.getError().getErrorSummary()); - if (isVerbose() && submission.getExceptionStackTrace() != null) { + if (isVerbose() && submission.getError().getErrorDetails() != null) { print("@|bold Stack trace: |@"); - println(submission.getExceptionStackTrace()); + println(submission.getError().getErrorDetails()); } } } http://git-wip-us.apache.org/repos/asf/sqoop/blob/9151d305/submission/mapreduce/src/main/java/org/apache/sqoop/submission/mapreduce/MapreduceSubmissionEngine.java ---------------------------------------------------------------------- diff --git a/submission/mapreduce/src/main/java/org/apache/sqoop/submission/mapreduce/MapreduceSubmissionEngine.java b/submission/mapreduce/src/main/java/org/apache/sqoop/submission/mapreduce/MapreduceSubmissionEngine.java index 631ceca..e04c888 100644 --- a/submission/mapreduce/src/main/java/org/apache/sqoop/submission/mapreduce/MapreduceSubmissionEngine.java +++ b/submission/mapreduce/src/main/java/org/apache/sqoop/submission/mapreduce/MapreduceSubmissionEngine.java @@ -20,6 +20,8 @@ package org.apache.sqoop.submission.mapreduce; import java.io.File; import java.io.FilenameFilter; import java.io.IOException; +import java.io.PrintWriter; +import java.io.StringWriter; import java.net.MalformedURLException; import java.util.Map; @@ -41,6 +43,7 @@ import org.apache.sqoop.execution.mapreduce.MapreduceExecutionEngine; import org.apache.sqoop.driver.JobRequest; import org.apache.sqoop.job.MRJobConstants; import org.apache.sqoop.job.mr.MRConfigurationUtils; +import org.apache.sqoop.model.SubmissionError; import org.apache.sqoop.submission.SubmissionStatus; import org.apache.sqoop.submission.counter.Counter; import org.apache.sqoop.submission.counter.CounterGroup; @@ -210,8 +213,8 @@ public class MapreduceSubmissionEngine extends SubmissionEngine { MRConfigurationUtils.setConnectorJobConfig(Direction.TO, job, request.getJobConfig(Direction.TO)); MRConfigurationUtils.setDriverConfig(job, request.getDriverConfig()); - MRConfigurationUtils.setConnectorSchema(Direction.FROM, job, request.getSummary().getFromSchema()); - MRConfigurationUtils.setConnectorSchema(Direction.TO, job, request.getSummary().getToSchema()); + MRConfigurationUtils.setConnectorSchema(Direction.FROM, job, request.getJobSubmission().getFromSchema()); + MRConfigurationUtils.setConnectorSchema(Direction.TO, job, request.getJobSubmission().getToSchema()); if(request.getJobName() != null) { job.setJobName("Sqoop: " + request.getJobName()); @@ -247,12 +250,19 @@ public class MapreduceSubmissionEngine extends SubmissionEngine { } String jobId = job.getJobID().toString(); - request.getSummary().setExternalId(jobId); - request.getSummary().setExternalLink(job.getTrackingURL()); + request.getJobSubmission().setExternalJobId(jobId); + request.getJobSubmission().setExternalLink(job.getTrackingURL()); LOG.debug("Executed new map-reduce job with id " + jobId); } catch (Exception e) { - request.getSummary().setException(e); + SubmissionError error = new SubmissionError(); + error.setErrorSummary(e.toString()); + StringWriter writer = new StringWriter(); + e.printStackTrace(new PrintWriter(writer)); + writer.flush(); + error.setErrorDetails(writer.toString()); + + request.getJobSubmission().setError(error); LOG.error("Error in submitting job", e); return false; } @@ -263,9 +273,9 @@ public class MapreduceSubmissionEngine extends SubmissionEngine { * {@inheritDoc} */ @Override - public void stop(String submissionId) { + public void stop(String externalJobId) { try { - RunningJob runningJob = jobClient.getJob(JobID.forName(submissionId)); + RunningJob runningJob = jobClient.getJob(JobID.forName(externalJobId)); if(runningJob == null) { return; } @@ -280,9 +290,9 @@ public class MapreduceSubmissionEngine extends SubmissionEngine { * {@inheritDoc} */ @Override - public SubmissionStatus status(String submissionId) { + public SubmissionStatus status(String externalJobId) { try { - RunningJob runningJob = jobClient.getJob(JobID.forName(submissionId)); + RunningJob runningJob = jobClient.getJob(JobID.forName(externalJobId)); if(runningJob == null) { return SubmissionStatus.UNKNOWN; } @@ -295,18 +305,43 @@ public class MapreduceSubmissionEngine extends SubmissionEngine { } } + /** * {@inheritDoc} */ @Override - public double progress(String submissionId) { + public SubmissionError error(String externalJobId) { + try { + RunningJob runningJob = jobClient.getJob(JobID.forName(externalJobId)); + if (runningJob == null) { + return null; + } + int status = runningJob.getJobState(); + if (status == JobStatus.FAILED || status == JobStatus.KILLED) { + SubmissionError error = new SubmissionError(); + error.setErrorSummary("Job Failed with status:" + status); + error.setErrorDetails(runningJob.getFailureInfo()); + return error; + } + + } catch (IOException e) { + throw new SqoopException(MapreduceSubmissionError.MAPREDUCE_0003, e); + } + return null; + } + + /** + * {@inheritDoc} + */ + @Override + public double progress(String externalJobId) { try { // Get some reasonable approximation of map-reduce job progress // TODO(jarcec): What if we're running without reducers? - RunningJob runningJob = jobClient.getJob(JobID.forName(submissionId)); + RunningJob runningJob = jobClient.getJob(JobID.forName(externalJobId)); if(runningJob == null) { // Return default value - return super.progress(submissionId); + return super.progress(externalJobId); } return (runningJob.mapProgress() + runningJob.reduceProgress()) / 2; @@ -319,12 +354,12 @@ public class MapreduceSubmissionEngine extends SubmissionEngine { * {@inheritDoc} */ @Override - public Counters counters(String submissionId) { + public Counters counters(String externalJobId) { try { - RunningJob runningJob = jobClient.getJob(JobID.forName(submissionId)); + RunningJob runningJob = jobClient.getJob(JobID.forName(externalJobId)); if(runningJob == null) { // Return default value - return super.counters(submissionId); + return super.counters(externalJobId); } return convertMapreduceCounters(runningJob.getCounters()); @@ -337,9 +372,9 @@ public class MapreduceSubmissionEngine extends SubmissionEngine { * {@inheritDoc} */ @Override - public String externalLink(String submissionId) { + public String externalLink(String externalJobId) { try { - RunningJob runningJob = jobClient.getJob(JobID.forName(submissionId)); + RunningJob runningJob = jobClient.getJob(JobID.forName(externalJobId)); if(runningJob == null) { return null; } @@ -374,6 +409,7 @@ public class MapreduceSubmissionEngine extends SubmissionEngine { "Unknown status " + status); } + /** * Convert Hadoop counters to Sqoop counters. * http://git-wip-us.apache.org/repos/asf/sqoop/blob/9151d305/test/src/main/java/org/apache/sqoop/test/testcases/ConnectorTestCase.java ---------------------------------------------------------------------- diff --git a/test/src/main/java/org/apache/sqoop/test/testcases/ConnectorTestCase.java b/test/src/main/java/org/apache/sqoop/test/testcases/ConnectorTestCase.java index 4f5f52a..a423785 100644 --- a/test/src/main/java/org/apache/sqoop/test/testcases/ConnectorTestCase.java +++ b/test/src/main/java/org/apache/sqoop/test/testcases/ConnectorTestCase.java @@ -246,11 +246,10 @@ abstract public class ConnectorTestCase extends TomcatTestCase { MSubmission finalSubmission = getClient().startJob(jid, DEFAULT_SUBMISSION_CALLBACKS, 100); if(finalSubmission.getStatus().isFailure()) { - LOG.error("Submission has failed: " + finalSubmission.getExceptionInfo()); - LOG.error("Corresponding stack trace: " + finalSubmission.getExceptionStackTrace()); + LOG.error("Submission has failed: " + finalSubmission.getError().getErrorSummary()); + LOG.error("Corresponding error details: " + finalSubmission.getError().getErrorDetails()); } - - assertEquals("Submission has failed with " + finalSubmission.getExceptionInfo(), SubmissionStatus.SUCCEEDED, finalSubmission.getStatus()); + assertEquals("Submission has failed with " + finalSubmission.getError().getErrorSummary(), SubmissionStatus.SUCCEEDED, finalSubmission.getStatus()); } /**
