Repository: sqoop Updated Branches: refs/heads/sqoop2 2f0e73067 -> 02786f0e2
SQOOP-1624: Sqoop2: Apply repository upgrader api Add supported direction check when upgrading connector as well. (Qian Xu 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/02786f0e Tree: http://git-wip-us.apache.org/repos/asf/sqoop/tree/02786f0e Diff: http://git-wip-us.apache.org/repos/asf/sqoop/diff/02786f0e Branch: refs/heads/sqoop2 Commit: 02786f0e205cab449b5cbceda30e288d2d6cce2d Parents: 2f0e730 Author: Abraham Elmahrek <[email protected]> Authored: Thu Dec 25 21:52:55 2014 -0800 Committer: Abraham Elmahrek <[email protected]> Committed: Thu Dec 25 21:57:36 2014 -0800 ---------------------------------------------------------------------- .../sqoop/connector/kite/KiteConnector.java | 3 +- .../connector/kite/KiteConnectorUpgrader.java | 47 ++++++++++ .../org/apache/sqoop/repository/Repository.java | 91 +++++++++++--------- 3 files changed, 96 insertions(+), 45 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/sqoop/blob/02786f0e/connector/connector-kite/src/main/java/org/apache/sqoop/connector/kite/KiteConnector.java ---------------------------------------------------------------------- diff --git a/connector/connector-kite/src/main/java/org/apache/sqoop/connector/kite/KiteConnector.java b/connector/connector-kite/src/main/java/org/apache/sqoop/connector/kite/KiteConnector.java index c864882..982d6dd 100644 --- a/connector/connector-kite/src/main/java/org/apache/sqoop/connector/kite/KiteConnector.java +++ b/connector/connector-kite/src/main/java/org/apache/sqoop/connector/kite/KiteConnector.java @@ -92,8 +92,7 @@ public class KiteConnector extends SqoopConnector { @Override public ConnectorConfigurableUpgrader getConfigurableUpgrader() { - // TODO: SQOOP-1624 - return null; + return new KiteConnectorUpgrader(); } } \ No newline at end of file http://git-wip-us.apache.org/repos/asf/sqoop/blob/02786f0e/connector/connector-kite/src/main/java/org/apache/sqoop/connector/kite/KiteConnectorUpgrader.java ---------------------------------------------------------------------- diff --git a/connector/connector-kite/src/main/java/org/apache/sqoop/connector/kite/KiteConnectorUpgrader.java b/connector/connector-kite/src/main/java/org/apache/sqoop/connector/kite/KiteConnectorUpgrader.java new file mode 100644 index 0000000..d3b9f95 --- /dev/null +++ b/connector/connector-kite/src/main/java/org/apache/sqoop/connector/kite/KiteConnectorUpgrader.java @@ -0,0 +1,47 @@ +/* + * 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.connector.kite; + +import org.apache.log4j.Logger; +import org.apache.sqoop.configurable.ConfigurableUpgradeUtil; +import org.apache.sqoop.connector.spi.ConnectorConfigurableUpgrader; +import org.apache.sqoop.model.MConfig; +import org.apache.sqoop.model.MFromConfig; +import org.apache.sqoop.model.MLinkConfig; +import org.apache.sqoop.model.MToConfig; + +//NOTE: All config types have the similar upgrade path at this point +public class KiteConnectorUpgrader extends ConnectorConfigurableUpgrader { + + @Override + public void upgradeLinkConfig(MLinkConfig original, MLinkConfig upgradeTarget) { + ConfigurableUpgradeUtil.doUpgrade(original.getConfigs(), upgradeTarget.getConfigs()); + } + + @Override + public void upgradeFromJobConfig(MFromConfig original, MFromConfig upgradeTarget) { + ConfigurableUpgradeUtil.doUpgrade(original.getConfigs(), upgradeTarget.getConfigs()); + } + + @Override + public void upgradeToJobConfig(MToConfig original, MToConfig upgradeTarget) { + ConfigurableUpgradeUtil.doUpgrade(original.getConfigs(), upgradeTarget.getConfigs()); + } + +} http://git-wip-us.apache.org/repos/asf/sqoop/blob/02786f0e/core/src/main/java/org/apache/sqoop/repository/Repository.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/sqoop/repository/Repository.java b/core/src/main/java/org/apache/sqoop/repository/Repository.java index 9461afe..1e8350c 100644 --- a/core/src/main/java/org/apache/sqoop/repository/Repository.java +++ b/core/src/main/java/org/apache/sqoop/repository/Repository.java @@ -471,52 +471,57 @@ public abstract class Repository { // every job has 2 parts, the FROM and the TO links and their // corresponding connectors. LOG.info(" Job upgrade for job:" + job.getName()+ " for connector:" + connectorName); - List<MConfig> fromConfig = newConnector.getFromConfig().clone(false).getConfigs(); - if (job.getFromConnectorId() == newConnector.getPersistenceId()) { - MFromConfig newFromConfig = new MFromConfig(fromConfig); - MFromConfig oldFromCOnfig = job.getFromJobConfig(); - upgrader.upgradeFromJobConfig(oldFromCOnfig, newFromConfig); - MToConfig oldToConfig = job.getToJobConfig(); - // create a job with new FROM direction configs but old TO direction - // configs - MJob newJob = new MJob(job, newFromConfig, oldToConfig, job.getDriverConfig()); - - ConfigValidationResult validationResult = ConfigUtils.validateConfigs( - newJob.getFromJobConfig().getConfigs(), - connector.getJobConfigurationClass(Direction.FROM) - ); - - if(validationResult.getStatus().canProceed()) { - updateJob(newJob, tx); - } else { - logInvalidModelObject("fromJob", newJob, validationResult); - upgradeSuccessful = false; - LOG.error(" From JOB config upgrade FAILED for job: " + job.getName() + " for connector:" + connectorName); + + if (newConnector.getSupportedDirections().isDirectionSupported(Direction.FROM)) { + List<MConfig> fromConfig = newConnector.getFromConfig().clone(false).getConfigs(); + if (job.getFromConnectorId() == newConnector.getPersistenceId()) { + MFromConfig newFromConfig = new MFromConfig(fromConfig); + MFromConfig oldFromConfig = job.getFromJobConfig(); + upgrader.upgradeFromJobConfig(oldFromConfig, newFromConfig); + MToConfig oldToConfig = job.getToJobConfig(); + // create a job with new FROM direction configs but old TO direction + // configs + MJob newJob = new MJob(job, newFromConfig, oldToConfig, job.getDriverConfig()); + + ConfigValidationResult validationResult = ConfigUtils.validateConfigs( + newJob.getFromJobConfig().getConfigs(), + connector.getJobConfigurationClass(Direction.FROM) + ); + + if (validationResult.getStatus().canProceed()) { + updateJob(newJob, tx); + } else { + logInvalidModelObject("fromJob", newJob, validationResult); + upgradeSuccessful = false; + LOG.error(" From JOB config upgrade FAILED for job: " + job.getName() + " for connector:" + connectorName); + } } } - List<MConfig> toConfig = newConnector.getToConfig().clone(false).getConfigs(); - if (job.getToConnectorId() == newConnector.getPersistenceId()) { - MToConfig oldToConfig = job.getToJobConfig(); - MToConfig newToConfig = new MToConfig(toConfig); - upgrader.upgradeToJobConfig(oldToConfig, newToConfig); - MFromConfig oldFromConfig = job.getFromJobConfig(); - // create a job with old FROM direction configs but new TO direction - // configs - MJob newJob = new MJob(job, oldFromConfig, newToConfig, job.getDriverConfig()); - - ConfigValidationResult validationResult = ConfigUtils.validateConfigs( - newJob.getToJobConfig().getConfigs(), - connector.getJobConfigurationClass(Direction.TO) - ); - - if(validationResult.getStatus().canProceed()) { - updateJob(newJob, tx); - } else { - logInvalidModelObject("toJob", newJob, validationResult); - upgradeSuccessful = false; - LOG.error(" TO JOB config upgrade FAILED for job: " + job.getName() + " for connector:" + connectorName); - } + if (newConnector.getSupportedDirections().isDirectionSupported(Direction.TO)) { + List<MConfig> toConfig = newConnector.getToConfig().clone(false).getConfigs(); + if (job.getToConnectorId() == newConnector.getPersistenceId()) { + MToConfig oldToConfig = job.getToJobConfig(); + MToConfig newToConfig = new MToConfig(toConfig); + upgrader.upgradeToJobConfig(oldToConfig, newToConfig); + MFromConfig oldFromConfig = job.getFromJobConfig(); + // create a job with old FROM direction configs but new TO direction + // configs + MJob newJob = new MJob(job, oldFromConfig, newToConfig, job.getDriverConfig()); + + ConfigValidationResult validationResult = ConfigUtils.validateConfigs( + newJob.getToJobConfig().getConfigs(), + connector.getJobConfigurationClass(Direction.TO) + ); + + if (validationResult.getStatus().canProceed()) { + updateJob(newJob, tx); + } else { + logInvalidModelObject("toJob", newJob, validationResult); + upgradeSuccessful = false; + LOG.error(" TO JOB config upgrade FAILED for job: " + job.getName() + " for connector:" + connectorName); + } + } } } }
