Repository: sqoop Updated Branches: refs/heads/sqoop2 788fd54a3 -> d013d94db
SQOOP-2360: Sqoop2: Upgrade should consider both directions for a single connector (Abraham Elmahrek via Jarek Jarcec Cecho) Project: http://git-wip-us.apache.org/repos/asf/sqoop/repo Commit: http://git-wip-us.apache.org/repos/asf/sqoop/commit/d013d94d Tree: http://git-wip-us.apache.org/repos/asf/sqoop/tree/d013d94d Diff: http://git-wip-us.apache.org/repos/asf/sqoop/diff/d013d94d Branch: refs/heads/sqoop2 Commit: d013d94dbb9a782f66d9e2aba4ee809db0ece024 Parents: 788fd54 Author: Jarek Jarcec Cecho <[email protected]> Authored: Tue May 26 12:29:20 2015 -0700 Committer: Jarek Jarcec Cecho <[email protected]> Committed: Tue May 26 12:29:20 2015 -0700 ---------------------------------------------------------------------- .../org/apache/sqoop/repository/Repository.java | 119 +++++++++++-------- .../sqoop/repository/TestJdbcRepository.java | 2 +- .../derby/upgrade/Derby1_99_4UpgradeTest.java | 5 +- .../derby/derby-repository-1.99.4.tar.gz | Bin 198069 -> 330779 bytes 4 files changed, 75 insertions(+), 51 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/sqoop/blob/d013d94d/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 aa91661..8cbff99 100644 --- a/core/src/main/java/org/apache/sqoop/repository/Repository.java +++ b/core/src/main/java/org/apache/sqoop/repository/Repository.java @@ -25,6 +25,7 @@ import org.apache.commons.lang.StringUtils; import org.apache.log4j.Logger; import org.apache.sqoop.common.Direction; import org.apache.sqoop.common.SqoopException; +import org.apache.sqoop.common.SupportedDirections; import org.apache.sqoop.connector.ConnectorManager; import org.apache.sqoop.connector.spi.ConnectorConfigurableUpgrader; import org.apache.sqoop.connector.spi.SqoopConnector; @@ -547,55 +548,77 @@ public abstract class Repository { // corresponding connectors. LOG.info(" Job upgrade 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); - } + SupportedDirections supportedDirections = newConnector.getSupportedDirections(); + + if (supportedDirections.isDirectionSupported(Direction.FROM) + && job.getFromConnectorId() == newConnector.getPersistenceId() + && supportedDirections.isDirectionSupported(Direction.TO) + && job.getToConnectorId() == newConnector.getPersistenceId()) { + // Upgrade both configs + MFromConfig newFromConfig = new MFromConfig(newConnector.getFromConfig().clone(false).getConfigs()); + MFromConfig oldFromConfig = job.getFromJobConfig(); + MToConfig newToConfig = new MToConfig(newConnector.getToConfig().clone(false).getConfigs()); + MToConfig oldToConfig = job.getToJobConfig(); + upgrader.upgradeFromJobConfig(oldFromConfig, newFromConfig); + upgrader.upgradeToJobConfig(oldToConfig, newToConfig); + + MJob newJob = new MJob(job, newFromConfig, newToConfig, job.getDriverConfig()); + + ConfigValidationResult validationResult = ConfigUtils.validateConfigs( + newJob.getFromJobConfig().getConfigs(), + connector.getJobConfigurationClass(Direction.FROM) + ); + + if (validationResult.getStatus().canProceed()) { + updateJob(newJob, tx); + } else { + logInvalidModelObject("job", newJob, validationResult); + upgradeSuccessful = false; + LOG.error(" 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); - } + } else if (supportedDirections.isDirectionSupported(Direction.FROM) + && job.getFromConnectorId() == newConnector.getPersistenceId()) { + MFromConfig newFromConfig = new MFromConfig(newConnector.getFromConfig().clone(false).getConfigs()); + 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("job", newJob, validationResult); + upgradeSuccessful = false; + LOG.error(" FROM JOB config upgrade FAILED for job: " + job.getName() + " for connector:" + connectorName); + } + } else if (supportedDirections.isDirectionSupported(Direction.TO) + && job.getToConnectorId() == newConnector.getPersistenceId()) { + MToConfig oldToConfig = job.getToJobConfig(); + MToConfig newToConfig = new MToConfig(newConnector.getToConfig().clone(false).getConfigs()); + 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("job", newJob, validationResult); + upgradeSuccessful = false; + LOG.error(" TO JOB config upgrade FAILED for job: " + job.getName() + " for connector:" + connectorName); } } } http://git-wip-us.apache.org/repos/asf/sqoop/blob/d013d94d/core/src/test/java/org/apache/sqoop/repository/TestJdbcRepository.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/sqoop/repository/TestJdbcRepository.java b/core/src/test/java/org/apache/sqoop/repository/TestJdbcRepository.java index 9b153fc..f571c80 100644 --- a/core/src/test/java/org/apache/sqoop/repository/TestJdbcRepository.java +++ b/core/src/test/java/org/apache/sqoop/repository/TestJdbcRepository.java @@ -245,7 +245,7 @@ public class TestJdbcRepository { repoOrder.verify(repoSpy, times(1)).deleteLinkInputs(2, repoTransactionMock); repoOrder.verify(repoSpy, times(1)).upgradeConnectorAndConfigs(any(MConnector.class), any(RepositoryTransaction.class)); repoOrder.verify(repoSpy, times(2)).updateLink(any(MLink.class), any(RepositoryTransaction.class)); - repoOrder.verify(repoSpy, times(4)).updateJob(any(MJob.class), any(RepositoryTransaction.class)); + repoOrder.verify(repoSpy, times(2)).updateJob(any(MJob.class), any(RepositoryTransaction.class)); repoOrder.verifyNoMoreInteractions(); txOrder.verify(repoTransactionMock, times(1)).begin(); txOrder.verify(repoTransactionMock, times(1)).commit(); http://git-wip-us.apache.org/repos/asf/sqoop/blob/d013d94d/test/src/test/java/org/apache/sqoop/integration/repository/derby/upgrade/Derby1_99_4UpgradeTest.java ---------------------------------------------------------------------- diff --git a/test/src/test/java/org/apache/sqoop/integration/repository/derby/upgrade/Derby1_99_4UpgradeTest.java b/test/src/test/java/org/apache/sqoop/integration/repository/derby/upgrade/Derby1_99_4UpgradeTest.java index 85ce8ff..9ee0379 100644 --- a/test/src/test/java/org/apache/sqoop/integration/repository/derby/upgrade/Derby1_99_4UpgradeTest.java +++ b/test/src/test/java/org/apache/sqoop/integration/repository/derby/upgrade/Derby1_99_4UpgradeTest.java @@ -33,6 +33,7 @@ import java.util.Map; * Job (-f 3 -t 1) with name "Export" and id 3 * Job (-f 3 -t 1) with blank name and id 4 * Job (-f 3 -t 1) with blank name and id 5 + * Job (-f 1 -t 1) with name "SameConnector" and id 6 * Job with id 1 has been executed 3 times * Job with id 2 has been executed 3 times * Job with id 3 has been executed 1 times @@ -55,7 +56,7 @@ public class Derby1_99_4UpgradeTest extends DerbyRepositoryUpgradeTest { @Override public int getNumberOfJobs() { - return 5; + return 6; } @Override @@ -86,6 +87,6 @@ public class Derby1_99_4UpgradeTest extends DerbyRepositoryUpgradeTest { @Override public Integer[] getDeleteJobIds() { - return new Integer[] {1, 2, 3, 4, 5}; + return new Integer[] {1, 2, 3, 4, 5, 6}; } } http://git-wip-us.apache.org/repos/asf/sqoop/blob/d013d94d/test/src/test/resources/repository/derby/derby-repository-1.99.4.tar.gz ---------------------------------------------------------------------- diff --git a/test/src/test/resources/repository/derby/derby-repository-1.99.4.tar.gz b/test/src/test/resources/repository/derby/derby-repository-1.99.4.tar.gz index 7a6ceed..5912e7c 100644 Binary files a/test/src/test/resources/repository/derby/derby-repository-1.99.4.tar.gz and b/test/src/test/resources/repository/derby/derby-repository-1.99.4.tar.gz differ
