Repository: sqoop Updated Branches: refs/heads/sqoop2 3257b3855 -> 39a220000
http://git-wip-us.apache.org/repos/asf/sqoop/blob/39a22000/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 34bd8a5..ff9e0c3 100644 --- a/core/src/test/java/org/apache/sqoop/repository/TestJdbcRepository.java +++ b/core/src/test/java/org/apache/sqoop/repository/TestJdbcRepository.java @@ -43,14 +43,14 @@ import java.util.List; import org.apache.sqoop.common.Direction; import org.apache.sqoop.common.SqoopException; import org.apache.sqoop.connector.ConnectorManager; -import org.apache.sqoop.connector.spi.RepositoryUpgrader; +import org.apache.sqoop.connector.spi.ConnectorConfigurableUpgrader; import org.apache.sqoop.connector.spi.SqoopConnector; import org.apache.sqoop.driver.Driver; +import org.apache.sqoop.driver.DriverUpgrader; import org.apache.sqoop.json.DriverBean; import org.apache.sqoop.model.ConfigUtils; import org.apache.sqoop.model.ConfigurationClass; import org.apache.sqoop.model.MConfig; -import org.apache.sqoop.model.MConfigList; import org.apache.sqoop.model.MConnector; import org.apache.sqoop.model.MDriver; import org.apache.sqoop.model.MDriverConfig; @@ -65,7 +65,6 @@ import org.apache.sqoop.validation.Validator; import org.junit.Before; import org.junit.Test; import org.mockito.InOrder; -import org.mockito.Mockito; public class TestJdbcRepository { @@ -75,7 +74,8 @@ public class TestJdbcRepository { private Driver driverMock; private JdbcRepositoryHandler repoHandlerMock; private Validator validatorMock; - private RepositoryUpgrader upgraderMock; + private ConnectorConfigurableUpgrader connectorUpgraderMock; + private DriverUpgrader driverUpgraderMock; private ConfigValidator validRepoMock; private ConfigValidator invalidRepoMock; @@ -87,7 +87,8 @@ public class TestJdbcRepository { driverMock = mock(Driver.class); repoHandlerMock = mock(JdbcRepositoryHandler.class); validatorMock = mock(Validator.class); - upgraderMock = mock(RepositoryUpgrader.class); + connectorUpgraderMock = mock(ConnectorConfigurableUpgrader.class); + driverUpgraderMock = mock(DriverUpgrader.class); repoSpy = spy(new JdbcRepository(repoHandlerMock, null)); // setup transaction and connector manager @@ -100,8 +101,15 @@ public class TestJdbcRepository { invalidRepoMock = mock(ConfigValidator.class); when(invalidRepoMock.getStatus()).thenReturn(Status.UNACCEPTABLE); - doNothing().when(upgraderMock).upgrade(any(MLinkConfig.class), any(MLinkConfig.class)); - doNothing().when(upgraderMock).upgrade(any(MFromConfig.class), any(MFromConfig.class)); + doNothing().when(connectorUpgraderMock).upgradeLinkConfig(any(MLinkConfig.class), + any(MLinkConfig.class)); + doNothing().when(connectorUpgraderMock).upgradeFromJobConfig(any(MFromConfig.class), + any(MFromConfig.class)); + doNothing().when(connectorUpgraderMock).upgradeToJobConfig(any(MToConfig.class), + any(MToConfig.class)); + doNothing().when(driverUpgraderMock).upgradeJobConfig(any(MDriverConfig.class), + any(MDriverConfig.class)); + } /** @@ -117,14 +125,14 @@ public class TestJdbcRepository { // make the upgradeConnector to throw an exception to prove that it has been called SqoopException exception = new SqoopException(RepositoryError.JDBCREPO_0000, "upgradeConnector() has been called."); - doThrow(exception).when(connectorMgrMock).getConnector(anyString()); + doThrow(exception).when(connectorMgrMock).getSqoopConnector(anyString()); try { repoSpy.registerConnector(newConnector, true); } catch (SqoopException ex) { assertEquals(ex.getMessage(), exception.getMessage()); verify(repoHandlerMock, times(1)).findConnector(anyString(), any(Connection.class)); - verify(connectorMgrMock, times(1)).getConnector(anyString()); + verify(connectorMgrMock, times(1)).getSqoopConnector(anyString()); verifyNoMoreInteractions(repoHandlerMock); return ; } @@ -218,28 +226,29 @@ public class TestJdbcRepository { when(validatorMock.validateConfigForLink(any(MLink.class))).thenReturn(validRepoMock); when(validatorMock.validateConfigForJob(any(MJob.class))).thenReturn(validRepoMock); when(sqconnector.getConfigValidator()).thenReturn(validatorMock); - when(sqconnector.getRepositoryUpgrader()).thenReturn(upgraderMock); + when(sqconnector.getConfigurableUpgrader()).thenReturn(connectorUpgraderMock); when(sqconnector.getLinkConfigurationClass()).thenReturn(EmptyLinkConfiguration.class); when(sqconnector.getJobConfigurationClass(any(Direction.class))).thenReturn( EmptyJobConfiguration.class); - when(connectorMgrMock.getConnector(anyString())).thenReturn(sqconnector); + when(connectorMgrMock.getSqoopConnector(anyString())).thenReturn(sqconnector); // prepare the links and jobs + // the connector Id for both are the same List<MLink> linkList = links(link(1,1), link(2,1)); - List<MJob> jobList = jobs(job(1,1,1,1,1), job(2,1,1,2,1)); + List<MJob> jobList = jobs(job(1,1,1,1,1), job(2,1,1,2,2)); // mock necessary methods for upgradeConnector() procedure doReturn(linkList).when(repoSpy).findLinksForConnector(anyLong()); doReturn(jobList).when(repoSpy).findJobsForConnector(anyLong()); doNothing().when(repoSpy).updateLink(any(MLink.class), any(RepositoryTransaction.class)); doNothing().when(repoSpy).updateJob(any(MJob.class), any(RepositoryTransaction.class)); - doNothing().when(repoSpy).upgradeConnector(any(MConnector.class), any(RepositoryTransaction.class)); + doNothing().when(repoSpy).upgradeConnectorConfigs(any(MConnector.class), any(RepositoryTransaction.class)); repoSpy.upgradeConnector(oldConnector, newConnector); InOrder repoOrder = inOrder(repoSpy); InOrder txOrder = inOrder(repoTransactionMock); - InOrder upgraderOrder = inOrder(upgraderMock); + InOrder upgraderOrder = inOrder(connectorUpgraderMock); InOrder validatorOrder = inOrder(validatorMock); repoOrder.verify(repoSpy, times(1)).findLinksForConnector(anyLong()); @@ -249,7 +258,7 @@ public class TestJdbcRepository { repoOrder.verify(repoSpy, times(1)).deleteJobInputs(2, repoTransactionMock); repoOrder.verify(repoSpy, times(1)).deleteLinkInputs(1, repoTransactionMock); repoOrder.verify(repoSpy, times(1)).deleteLinkInputs(2, repoTransactionMock); - repoOrder.verify(repoSpy, times(1)).upgradeConnector(any(MConnector.class), any(RepositoryTransaction.class)); + repoOrder.verify(repoSpy, times(1)).upgradeConnectorConfigs(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.verifyNoMoreInteractions(); @@ -257,8 +266,11 @@ public class TestJdbcRepository { txOrder.verify(repoTransactionMock, times(1)).commit(); txOrder.verify(repoTransactionMock, times(1)).close(); txOrder.verifyNoMoreInteractions(); - upgraderOrder.verify(upgraderMock, times(2)).upgrade(any(MLinkConfig.class), any(MLinkConfig.class)); - upgraderOrder.verify(upgraderMock, times(4)).upgrade(any(MFromConfig.class), any(MFromConfig.class)); + upgraderOrder.verify(connectorUpgraderMock, times(2)).upgradeLinkConfig(any(MLinkConfig.class), any(MLinkConfig.class)); + upgraderOrder.verify(connectorUpgraderMock, times(1)).upgradeFromJobConfig(any(MFromConfig.class), any(MFromConfig.class)); + upgraderOrder.verify(connectorUpgraderMock, times(1)).upgradeToJobConfig(any(MToConfig.class), any(MToConfig.class)); + upgraderOrder.verify(connectorUpgraderMock, times(1)).upgradeFromJobConfig(any(MFromConfig.class), any(MFromConfig.class)); + upgraderOrder.verify(connectorUpgraderMock, times(1)).upgradeToJobConfig(any(MToConfig.class), any(MToConfig.class)); upgraderOrder.verifyNoMoreInteractions(); validatorOrder.verify(validatorMock, times(2)).validateConfigForLink(anyObject()); // @TODO(Abe): Re-enable job validation? @@ -277,34 +289,34 @@ public class TestJdbcRepository { when(validatorMock.validateConfigForLink(any(MLink.class))).thenReturn(validRepoMock); when(validatorMock.validateConfigForJob(any(MJob.class))).thenReturn(validRepoMock); when(driverMock.getValidator()).thenReturn(validatorMock); - when(driverMock.getDriverConfigRepositoryUpgrader()).thenReturn(upgraderMock); - when(driverMock.getDriverConfigurationGroupClass()).thenReturn(EmptyJobConfiguration.class); + when(driverMock.getConfigurableUpgrader()).thenReturn(driverUpgraderMock); + when(driverMock.getDriverJobConfigurationClass()).thenReturn(EmptyJobConfiguration.class); List<MJob> jobList = jobs(job(1,1,1,1,1), job(2,1,1,2,1)); doReturn(jobList).when(repoSpy).findJobs(); doNothing().when(repoSpy).updateLink(any(MLink.class), any(RepositoryTransaction.class)); doNothing().when(repoSpy).updateJob(any(MJob.class), any(RepositoryTransaction.class)); - doNothing().when(repoSpy).upgradeDriver(any(MDriver.class), any(RepositoryTransaction.class)); + doNothing().when(repoSpy).upgradeDriverConfigs(any(MDriver.class), any(RepositoryTransaction.class)); repoSpy.upgradeDriver(newDriverConfig); InOrder repoOrder = inOrder(repoSpy); InOrder txOrder = inOrder(repoTransactionMock); - InOrder upgraderOrder = inOrder(upgraderMock); + InOrder upgraderOrder = inOrder(driverUpgraderMock); InOrder validatorOrder = inOrder(validatorMock); repoOrder.verify(repoSpy, times(1)).findJobs(); repoOrder.verify(repoSpy, times(1)).getTransaction(); repoOrder.verify(repoSpy, times(1)).deleteJobInputs(1, repoTransactionMock); repoOrder.verify(repoSpy, times(1)).deleteJobInputs(2, repoTransactionMock); - repoOrder.verify(repoSpy, times(1)).upgradeDriver(any(MDriver.class), any(RepositoryTransaction.class)); + repoOrder.verify(repoSpy, times(1)).upgradeDriverConfigs(any(MDriver.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(); txOrder.verify(repoTransactionMock, times(1)).close(); txOrder.verifyNoMoreInteractions(); - upgraderOrder.verify(upgraderMock, times(2)).upgrade(any(MConfigList.class), any(MConfigList.class)); + upgraderOrder.verify(driverUpgraderMock, times(2)).upgradeJobConfig(any(MDriverConfig.class), any(MDriverConfig.class)); upgraderOrder.verifyNoMoreInteractions(); validatorOrder.verify(validatorMock, times(2)).validateConfigForJob(anyObject()); validatorOrder.verifyNoMoreInteractions(); @@ -321,13 +333,13 @@ public class TestJdbcRepository { when(validatorMock.validateConfigForLink(any(MLink.class))).thenReturn(invalidRepoMock); when(validatorMock.validateConfigForJob(any(MJob.class))).thenReturn(invalidRepoMock); when(driverMock.getValidator()).thenReturn(validatorMock); - when(driverMock.getDriverConfigRepositoryUpgrader()).thenReturn(upgraderMock); - when(driverMock.getDriverConfigurationGroupClass()).thenReturn(EmptyJobConfiguration.class); + when(driverMock.getConfigurableUpgrader()).thenReturn(driverUpgraderMock); + when(driverMock.getDriverJobConfigurationClass()).thenReturn(EmptyJobConfiguration.class); List<MJob> jobList = jobs(job(1,1,1,1,1), job(2,1,1,2,1)); doReturn(jobList).when(repoSpy).findJobs(); doNothing().when(repoSpy).updateJob(any(MJob.class), any(RepositoryTransaction.class)); - doNothing().when(repoSpy).upgradeDriver(any(MDriver.class), any(RepositoryTransaction.class)); + doNothing().when(repoSpy).upgradeDriverConfigs(any(MDriver.class), any(RepositoryTransaction.class)); try { repoSpy.upgradeDriver(newDriverConfig); @@ -336,20 +348,20 @@ public class TestJdbcRepository { InOrder repoOrder = inOrder(repoSpy); InOrder txOrder = inOrder(repoTransactionMock); - InOrder upgraderOrder = inOrder(upgraderMock); + InOrder upgraderOrder = inOrder(driverUpgraderMock); InOrder validatorOrder = inOrder(validatorMock); repoOrder.verify(repoSpy, times(1)).findJobs(); repoOrder.verify(repoSpy, times(1)).getTransaction(); repoOrder.verify(repoSpy, times(1)).deleteJobInputs(1, repoTransactionMock); repoOrder.verify(repoSpy, times(1)).deleteJobInputs(2, repoTransactionMock); - repoOrder.verify(repoSpy, times(1)).upgradeDriver(any(MDriver.class), any(RepositoryTransaction.class)); + repoOrder.verify(repoSpy, times(1)).upgradeDriverConfigs(any(MDriver.class), any(RepositoryTransaction.class)); repoOrder.verifyNoMoreInteractions(); txOrder.verify(repoTransactionMock, times(1)).begin(); txOrder.verify(repoTransactionMock, times(1)).rollback(); txOrder.verify(repoTransactionMock, times(1)).close(); txOrder.verifyNoMoreInteractions(); - upgraderOrder.verify(upgraderMock, times(2)).upgrade(any(MConfigList.class), any(MConfigList.class)); + upgraderOrder.verify(driverUpgraderMock, times(2)).upgradeJobConfig(any(MDriverConfig.class), any(MDriverConfig.class)); upgraderOrder.verifyNoMoreInteractions(); // driver configs are per job. validatorOrder.verify(validatorMock, times(2)).validateConfigForJob(anyObject()); @@ -371,8 +383,8 @@ public class TestJdbcRepository { SqoopConnector sqconnector = mock(SqoopConnector.class); when(sqconnector.getConfigValidator()).thenReturn(validatorMock); - when(sqconnector.getRepositoryUpgrader()).thenReturn(upgraderMock); - when(connectorMgrMock.getConnector(anyString())).thenReturn(sqconnector); + when(sqconnector.getConfigurableUpgrader()).thenReturn(connectorUpgraderMock); + when(connectorMgrMock.getSqoopConnector(anyString())).thenReturn(sqconnector); SqoopException exception = new SqoopException(RepositoryError.JDBCREPO_0000, "find links for connector error."); @@ -401,8 +413,8 @@ public class TestJdbcRepository { SqoopConnector sqconnector = mock(SqoopConnector.class); when(sqconnector.getConfigValidator()).thenReturn(validatorMock); - when(sqconnector.getRepositoryUpgrader()).thenReturn(upgraderMock); - when(connectorMgrMock.getConnector(anyString())).thenReturn(sqconnector); + when(sqconnector.getConfigurableUpgrader()).thenReturn(connectorUpgraderMock); + when(connectorMgrMock.getSqoopConnector(anyString())).thenReturn(sqconnector); List<MLink> linkList = links(link(1,1), link(2,1)); doReturn(linkList).when(repoHandlerMock).findLinksForConnector(anyLong(), any(Connection.class)); @@ -435,8 +447,8 @@ public class TestJdbcRepository { SqoopConnector sqconnector = mock(SqoopConnector.class); when(sqconnector.getConfigValidator()).thenReturn(validatorMock); - when(sqconnector.getRepositoryUpgrader()).thenReturn(upgraderMock); - when(connectorMgrMock.getConnector(anyString())).thenReturn(sqconnector); + when(sqconnector.getConfigurableUpgrader()).thenReturn(connectorUpgraderMock); + when(connectorMgrMock.getSqoopConnector(anyString())).thenReturn(sqconnector); List<MLink> linkList = links(link(1,1), link(2,1)); List<MJob> jobList = jobs(job(1,1,1,1,1), job(2,1,1,2,1)); @@ -472,8 +484,8 @@ public class TestJdbcRepository { SqoopConnector sqconnector = mock(SqoopConnector.class); when(sqconnector.getConfigValidator()).thenReturn(validatorMock); - when(sqconnector.getRepositoryUpgrader()).thenReturn(upgraderMock); - when(connectorMgrMock.getConnector(anyString())).thenReturn(sqconnector); + when(sqconnector.getConfigurableUpgrader()).thenReturn(connectorUpgraderMock); + when(connectorMgrMock.getSqoopConnector(anyString())).thenReturn(sqconnector); List<MLink> linkList = links(link(1,1), link(2,1)); List<MJob> jobList = jobs(job(1,1,1,1,1), job(2,1,1,2,1)); @@ -511,8 +523,8 @@ public class TestJdbcRepository { SqoopConnector sqconnector = mock(SqoopConnector.class); when(sqconnector.getConfigValidator()).thenReturn(validatorMock); - when(sqconnector.getRepositoryUpgrader()).thenReturn(upgraderMock); - when(connectorMgrMock.getConnector(anyString())).thenReturn(sqconnector); + when(sqconnector.getConfigurableUpgrader()).thenReturn(connectorUpgraderMock); + when(connectorMgrMock.getSqoopConnector(anyString())).thenReturn(sqconnector); List<MLink> linkList = links(link(1,1), link(2,1)); List<MJob> jobList = jobs(job(1,1,1,1,1), job(2,1,1,2,1)); @@ -523,7 +535,7 @@ public class TestJdbcRepository { SqoopException exception = new SqoopException(RepositoryError.JDBCREPO_0000, "update connector error."); - doThrow(exception).when(repoHandlerMock).upgradeConnector(any(MConnector.class), any(Connection.class)); + doThrow(exception).when(repoHandlerMock).upgradeConnectorConfigs(any(MConnector.class), any(Connection.class)); try { repoSpy.upgradeConnector(oldConnector, newConnector); @@ -533,7 +545,7 @@ public class TestJdbcRepository { verify(repoHandlerMock, times(1)).findJobsForConnector(anyLong(), any(Connection.class)); verify(repoHandlerMock, times(2)).deleteJobInputs(anyLong(), any(Connection.class)); verify(repoHandlerMock, times(2)).deleteLinkInputs(anyLong(), any(Connection.class)); - verify(repoHandlerMock, times(1)).upgradeConnector(any(MConnector.class), any(Connection.class)); + verify(repoHandlerMock, times(1)).upgradeConnectorConfigs(any(MConnector.class), any(Connection.class)); verifyNoMoreInteractions(repoHandlerMock); return ; } @@ -554,10 +566,10 @@ public class TestJdbcRepository { when(validatorMock.validateConfigForLink(any(MLink.class))).thenReturn(validRepoMock); when(validatorMock.validateConfigForJob(any(MJob.class))).thenReturn(validRepoMock); when(sqconnector.getConfigValidator()).thenReturn(validatorMock); - when(sqconnector.getRepositoryUpgrader()).thenReturn(upgraderMock); + when(sqconnector.getConfigurableUpgrader()).thenReturn(connectorUpgraderMock); when(sqconnector.getLinkConfigurationClass()).thenReturn(EmptyLinkConfiguration.class); when(sqconnector.getJobConfigurationClass(any(Direction.class))).thenReturn(EmptyJobConfiguration.class); - when(connectorMgrMock.getConnector(anyString())).thenReturn(sqconnector); + when(connectorMgrMock.getSqoopConnector(anyString())).thenReturn(sqconnector); List<MLink> linkList = links(link(1,1), link(2,1)); List<MJob> jobList = jobs(job(1,1,1,1,1), job(2,1,1,2,1)); @@ -565,7 +577,7 @@ public class TestJdbcRepository { doReturn(jobList).when(repoHandlerMock).findJobsForConnector(anyLong(), any(Connection.class)); doNothing().when(repoHandlerMock).deleteJobInputs(anyLong(), any(Connection.class)); doNothing().when(repoHandlerMock).deleteLinkInputs(anyLong(), any(Connection.class)); - doNothing().when(repoHandlerMock).upgradeConnector(any(MConnector.class), any(Connection.class)); + doNothing().when(repoHandlerMock).upgradeConnectorConfigs(any(MConnector.class), any(Connection.class)); doReturn(true).when(repoHandlerMock).existsLink(anyLong(), any(Connection.class)); SqoopException exception = new SqoopException(RepositoryError.JDBCREPO_0000, @@ -580,7 +592,7 @@ public class TestJdbcRepository { verify(repoHandlerMock, times(1)).findJobsForConnector(anyLong(), any(Connection.class)); verify(repoHandlerMock, times(2)).deleteJobInputs(anyLong(), any(Connection.class)); verify(repoHandlerMock, times(2)).deleteLinkInputs(anyLong(), any(Connection.class)); - verify(repoHandlerMock, times(1)).upgradeConnector(any(MConnector.class), any(Connection.class)); + verify(repoHandlerMock, times(1)).upgradeConnectorConfigs(any(MConnector.class), any(Connection.class)); verify(repoHandlerMock, times(1)).existsLink(anyLong(), any(Connection.class)); verify(repoHandlerMock, times(1)).updateLink(any(MLink.class), any(Connection.class)); verifyNoMoreInteractions(repoHandlerMock); @@ -603,10 +615,10 @@ public class TestJdbcRepository { when(validatorMock.validateConfigForLink(any(MLink.class))).thenReturn(validRepoMock); when(validatorMock.validateConfigForJob(any(MJob.class))).thenReturn(validRepoMock); when(sqconnector.getConfigValidator()).thenReturn(validatorMock); - when(sqconnector.getRepositoryUpgrader()).thenReturn(upgraderMock); + when(sqconnector.getConfigurableUpgrader()).thenReturn(connectorUpgraderMock); when(sqconnector.getLinkConfigurationClass()).thenReturn(EmptyLinkConfiguration.class); when(sqconnector.getJobConfigurationClass(any(Direction.class))).thenReturn(EmptyJobConfiguration.class); - when(connectorMgrMock.getConnector(anyString())).thenReturn(sqconnector); + when(connectorMgrMock.getSqoopConnector(anyString())).thenReturn(sqconnector); List<MLink> linkList = links(link(1,1), link(2,1)); List<MJob> jobList = jobs(job(1,1,1,1,1), job(2,1,1,2,1)); @@ -614,7 +626,7 @@ public class TestJdbcRepository { doReturn(jobList).when(repoHandlerMock).findJobsForConnector(anyLong(), any(Connection.class)); doNothing().when(repoHandlerMock).deleteJobInputs(anyLong(), any(Connection.class)); doNothing().when(repoHandlerMock).deleteLinkInputs(anyLong(), any(Connection.class)); - doNothing().when(repoHandlerMock).upgradeConnector(any(MConnector.class), any(Connection.class)); + doNothing().when(repoHandlerMock).upgradeConnectorConfigs(any(MConnector.class), any(Connection.class)); doNothing().when(repoHandlerMock).updateLink(any(MLink.class), any(Connection.class)); doReturn(true).when(repoHandlerMock).existsLink(anyLong(), any(Connection.class)); doReturn(true).when(repoHandlerMock).existsJob(anyLong(), any(Connection.class)); @@ -631,7 +643,7 @@ public class TestJdbcRepository { verify(repoHandlerMock, times(1)).findJobsForConnector(anyLong(), any(Connection.class)); verify(repoHandlerMock, times(2)).deleteJobInputs(anyLong(), any(Connection.class)); verify(repoHandlerMock, times(2)).deleteLinkInputs(anyLong(), any(Connection.class)); - verify(repoHandlerMock, times(1)).upgradeConnector(any(MConnector.class), any(Connection.class)); + verify(repoHandlerMock, times(1)).upgradeConnectorConfigs(any(MConnector.class), any(Connection.class)); verify(repoHandlerMock, times(2)).existsLink(anyLong(), any(Connection.class)); verify(repoHandlerMock, times(2)).updateLink(any(MLink.class), any(Connection.class)); verify(repoHandlerMock, times(1)).existsJob(anyLong(), any(Connection.class)); @@ -652,7 +664,7 @@ public class TestJdbcRepository { MDriver newDriverConfig = driver(); when(driverMock.getValidator()).thenReturn(validatorMock); - when(driverMock.getDriverConfigRepositoryUpgrader()).thenReturn(upgraderMock); + when(driverMock.getConfigurableUpgrader()).thenReturn(driverUpgraderMock); SqoopException exception = new SqoopException(RepositoryError.JDBCREPO_0000, "find jobs error."); @@ -679,7 +691,7 @@ public class TestJdbcRepository { MDriver newDriverConfig = driver(); when(driverMock.getValidator()).thenReturn(validatorMock); - when(driverMock.getDriverConfigRepositoryUpgrader()).thenReturn(upgraderMock); + when(driverMock.getConfigurableUpgrader()).thenReturn(driverUpgraderMock); List<MJob> jobList = jobs(job(1,1,1,1,1), job(2,1,1,2,1)); doReturn(jobList).when(repoHandlerMock).findJobs(any(Connection.class)); @@ -710,7 +722,7 @@ public class TestJdbcRepository { MDriver newDriverConfig = driver(); when(driverMock.getValidator()).thenReturn(validatorMock); - when(driverMock.getDriverConfigRepositoryUpgrader()).thenReturn(upgraderMock); + when(driverMock.getConfigurableUpgrader()).thenReturn(driverUpgraderMock); List<MJob> jobList = jobs(job(1,1,1,1,1), job(2,1,1,2,1)); doReturn(jobList).when(repoHandlerMock).findJobs(any(Connection.class)); @@ -719,7 +731,7 @@ public class TestJdbcRepository { SqoopException exception = new SqoopException(RepositoryError.JDBCREPO_0000, "update driverConfig entity error."); - doThrow(exception).when(repoHandlerMock).upgradeDriver(any(MDriver.class), any(Connection.class)); + doThrow(exception).when(repoHandlerMock).upgradeDriverConfigs(any(MDriver.class), any(Connection.class)); try { repoSpy.upgradeDriver(newDriverConfig); @@ -727,7 +739,7 @@ public class TestJdbcRepository { assertEquals(ex.getMessage(), exception.getMessage()); verify(repoHandlerMock, times(1)).findJobs(any(Connection.class)); verify(repoHandlerMock, times(2)).deleteJobInputs(anyLong(), any(Connection.class)); - verify(repoHandlerMock, times(1)).upgradeDriver(any(MDriver.class), any(Connection.class)); + verify(repoHandlerMock, times(1)).upgradeDriverConfigs(any(MDriver.class), any(Connection.class)); verifyNoMoreInteractions(repoHandlerMock); return ; } @@ -747,12 +759,12 @@ public class TestJdbcRepository { when(validatorMock.validateConfigForLink(any(MLink.class))).thenReturn(validRepoMock); when(validatorMock.validateConfigForJob(any(MJob.class))).thenReturn(validRepoMock); when(driverMock.getValidator()).thenReturn(validatorMock); - when(driverMock.getDriverConfigRepositoryUpgrader()).thenReturn(upgraderMock); - when(driverMock.getDriverConfigurationGroupClass()).thenReturn(EmptyJobConfiguration.class); + when(driverMock.getConfigurableUpgrader()).thenReturn(driverUpgraderMock); + when(driverMock.getDriverJobConfigurationClass()).thenReturn(EmptyJobConfiguration.class); List<MJob> jobList = jobs(job(1,1,1,1,1), job(2,1,1,2,1)); doReturn(jobList).when(repoHandlerMock).findJobs(any(Connection.class)); doNothing().when(repoHandlerMock).deleteJobInputs(anyLong(), any(Connection.class)); - doNothing().when(repoHandlerMock).upgradeDriver(any(MDriver.class), any(Connection.class)); + doNothing().when(repoHandlerMock).upgradeDriverConfigs(any(MDriver.class), any(Connection.class)); doReturn(true).when(repoHandlerMock).existsJob(anyLong(), any(Connection.class)); SqoopException exception = new SqoopException(RepositoryError.JDBCREPO_0000, @@ -765,7 +777,7 @@ public class TestJdbcRepository { assertEquals(ex.getMessage(), exception.getMessage()); verify(repoHandlerMock, times(1)).findJobs(any(Connection.class)); verify(repoHandlerMock, times(2)).deleteJobInputs(anyLong(), any(Connection.class)); - verify(repoHandlerMock, times(1)).upgradeDriver(any(MDriver.class), any(Connection.class)); + verify(repoHandlerMock, times(1)).upgradeDriverConfigs(any(MDriver.class), any(Connection.class)); verify(repoHandlerMock, times(1)).existsJob(anyLong(), any(Connection.class)); verify(repoHandlerMock, times(1)).updateJob(any(MJob.class), any(Connection.class)); verifyNoMoreInteractions(repoHandlerMock); http://git-wip-us.apache.org/repos/asf/sqoop/blob/39a22000/repository/repository-derby/src/main/java/org/apache/sqoop/repository/derby/DerbyRepositoryHandler.java ---------------------------------------------------------------------- diff --git a/repository/repository-derby/src/main/java/org/apache/sqoop/repository/derby/DerbyRepositoryHandler.java b/repository/repository-derby/src/main/java/org/apache/sqoop/repository/derby/DerbyRepositoryHandler.java index c888910..aa58850 100644 --- a/repository/repository-derby/src/main/java/org/apache/sqoop/repository/derby/DerbyRepositoryHandler.java +++ b/repository/repository-derby/src/main/java/org/apache/sqoop/repository/derby/DerbyRepositoryHandler.java @@ -138,7 +138,7 @@ public class DerbyRepositoryHandler extends JdbcRepositoryHandler { * repository. The job and connector configs within <code>mc</code> will get * updated with the id of the configs when this function returns. * @param mc The connector to use for updating configs - * @param conn JDBC link to use for updating the configs + * @param conn JDBC connection to use for inserting the configs */ private void insertConfigsForConnector (MConnector mc, Connection conn) { long connectorId = mc.getPersistenceId(); @@ -151,17 +151,18 @@ public class DerbyRepositoryHandler extends JdbcRepositoryHandler { baseInputStmt = conn.prepareStatement(STMT_INSERT_INPUT_BASE, Statement.RETURN_GENERATED_KEYS); - // Register link type config + // Register link type config for connector + // NOTE: The direction is null for LINK type registerConfigs(connectorId, null, mc.getLinkConfig().getConfigs(), MConfigType.LINK.name(), baseConfigStmt, baseInputStmt, conn); - // Register both from/to job type config + // Register both from/to job type config for connector if (mc.getSupportedDirections().isDirectionSupported(Direction.FROM)) { - registerConfigs(connectorId, Direction.FROM, mc.getConfig(Direction.FROM).getConfigs(), + registerConfigs(connectorId, Direction.FROM, mc.getFromConfig().getConfigs(), MConfigType.JOB.name(), baseConfigStmt, baseInputStmt, conn); } if (mc.getSupportedDirections().isDirectionSupported(Direction.TO)) { - registerConfigs(connectorId, Direction.TO, mc.getConfig(Direction.TO).getConfigs(), + registerConfigs(connectorId, Direction.TO, mc.getToConfig().getConfigs(), MConfigType.JOB.name(), baseConfigStmt, baseInputStmt, conn); } } catch (SQLException ex) { @@ -752,8 +753,8 @@ public class DerbyRepositoryHandler extends JdbcRepositoryHandler { for (URL url : connectorConfigs) { handler = new ConnectorHandler(url); - if (handler.getMetadata().getPersistenceId() != -1) { - return handler.getMetadata().getPersistenceId(); + if (handler.getConnectorConfigurable().getPersistenceId() != -1) { + return handler.getConnectorConfigurable().getPersistenceId(); } if (handler.getUniqueName().equals(CONNECTOR_HDFS)) { @@ -761,8 +762,8 @@ public class DerbyRepositoryHandler extends JdbcRepositoryHandler { PreparedStatement baseConnectorStmt = conn.prepareStatement( STMT_INSERT_CONNECTOR_WITHOUT_SUPPORTED_DIRECTIONS, Statement.RETURN_GENERATED_KEYS); - baseConnectorStmt.setString(1, handler.getMetadata().getUniqueName()); - baseConnectorStmt.setString(2, handler.getMetadata().getClassName()); + baseConnectorStmt.setString(1, handler.getConnectorConfigurable().getUniqueName()); + baseConnectorStmt.setString(2, handler.getConnectorConfigurable().getClassName()); baseConnectorStmt.setString(3, "0"); if (baseConnectorStmt.executeUpdate() == 1) { ResultSet rsetConnectorId = baseConnectorStmt.getGeneratedKeys(); @@ -1229,9 +1230,7 @@ public class DerbyRepositoryHandler extends JdbcRepositoryHandler { try { stmt = conn.prepareStatement(STMT_SELECT_LINK_FOR_CONNECTOR); stmt.setLong(1, connectorID); - return loadLinks(stmt, conn); - } catch (SQLException ex) { logException(ex, connectorID); throw new SqoopException(DerbyRepoError.DERBYREPO_0023, ex); @@ -1244,7 +1243,12 @@ public class DerbyRepositoryHandler extends JdbcRepositoryHandler { * {@inheritDoc} */ @Override - public void upgradeConnector(MConnector mConnector, Connection conn) { + public void upgradeConnectorConfigs(MConnector mConnector, Connection conn) { + updateConnectorAndDeleteConfigs(mConnector, conn); + insertConfigsForConnector(mConnector, conn); + } + + private void updateConnectorAndDeleteConfigs(MConnector mConnector, Connection conn) { PreparedStatement updateConnectorStatement = null; PreparedStatement deleteConfig = null; PreparedStatement deleteInput = null; @@ -1271,15 +1275,19 @@ public class DerbyRepositoryHandler extends JdbcRepositoryHandler { } finally { closeStatements(updateConnectorStatement, deleteConfig, deleteInput); } - insertConfigsForConnector(mConnector, conn); - } /** * {@inheritDoc} */ @Override - public void upgradeDriver(MDriver mDriver, Connection conn) { + public void upgradeDriverConfigs(MDriver mDriver, Connection conn) { + updateDriverAndDeleteConfigs(mDriver, conn); + createOrUpdateDriverSystemVersion(conn, mDriver.getVersion()); + insertConfigsForDriver(mDriver, conn); + } + + private void updateDriverAndDeleteConfigs(MDriver mDriver, Connection conn) { PreparedStatement deleteConfig = null; PreparedStatement deleteInput = null; try { @@ -1295,8 +1303,6 @@ public class DerbyRepositoryHandler extends JdbcRepositoryHandler { } finally { closeStatements(deleteConfig, deleteInput); } - createOrUpdateDriverSystemVersion(conn, mDriver.getVersion()); - insertConfigsForDriver(mDriver, conn); } /** http://git-wip-us.apache.org/repos/asf/sqoop/blob/39a22000/repository/repository-derby/src/test/java/org/apache/sqoop/repository/derby/TestDriverHandling.java ---------------------------------------------------------------------- diff --git a/repository/repository-derby/src/test/java/org/apache/sqoop/repository/derby/TestDriverHandling.java b/repository/repository-derby/src/test/java/org/apache/sqoop/repository/derby/TestDriverHandling.java index 95fbe07..bbf721f 100644 --- a/repository/repository-derby/src/test/java/org/apache/sqoop/repository/derby/TestDriverHandling.java +++ b/repository/repository-derby/src/test/java/org/apache/sqoop/repository/derby/TestDriverHandling.java @@ -125,7 +125,7 @@ public class TestDriverHandling extends DerbyTestCase { + DerbyRepoConstants.SYSKEY_DRIVER_CONFIG_VERSION + "'"); assertEquals(lowerVersion, getDriverVersion()); - handler.upgradeDriver(driver, getDerbyDatabaseConnection()); + handler.upgradeDriverConfigs(driver, getDerbyDatabaseConnection()); assertEquals(CURRENT_DRIVER_VERSION, driver.getVersion()); http://git-wip-us.apache.org/repos/asf/sqoop/blob/39a22000/server/src/main/java/org/apache/sqoop/handler/ConnectorRequestHandler.java ---------------------------------------------------------------------- diff --git a/server/src/main/java/org/apache/sqoop/handler/ConnectorRequestHandler.java b/server/src/main/java/org/apache/sqoop/handler/ConnectorRequestHandler.java index 7109ae5..c50e029 100644 --- a/server/src/main/java/org/apache/sqoop/handler/ConnectorRequestHandler.java +++ b/server/src/main/java/org/apache/sqoop/handler/ConnectorRequestHandler.java @@ -70,7 +70,7 @@ public class ConnectorRequestHandler implements RequestHandler { LOG.info("ConnectorRequestHandler handles cid: " + cid); if (cid.equals("all")) { // display all connectors - connectors = ConnectorManager.getInstance().getConnectorsMetadata(); + connectors = ConnectorManager.getInstance().getConnectorConfigurables(); bundles = ConnectorManager.getInstance().getResourceBundles(locale); AuditLoggerManager.getInstance() @@ -87,7 +87,7 @@ public class ConnectorRequestHandler implements RequestHandler { connectors = new LinkedList<MConnector>(); bundles = new HashMap<Long, ResourceBundle>(); - connectors.add(ConnectorManager.getInstance().getConnectorConfig(id)); + connectors.add(ConnectorManager.getInstance().getConnectorConfigurable(id)); bundles.put(id, ConnectorManager.getInstance().getResourceBundle(id, locale)); AuditLoggerManager.getInstance() http://git-wip-us.apache.org/repos/asf/sqoop/blob/39a22000/server/src/main/java/org/apache/sqoop/handler/JobRequestHandler.java ---------------------------------------------------------------------- diff --git a/server/src/main/java/org/apache/sqoop/handler/JobRequestHandler.java b/server/src/main/java/org/apache/sqoop/handler/JobRequestHandler.java index 462579c..0cd5acb 100644 --- a/server/src/main/java/org/apache/sqoop/handler/JobRequestHandler.java +++ b/server/src/main/java/org/apache/sqoop/handler/JobRequestHandler.java @@ -164,10 +164,10 @@ public class JobRequestHandler implements RequestHandler { // Verify that user is not trying to spoof us MFromConfig fromConfig = ConnectorManager.getInstance() - .getConnectorConfig(job.getConnectorId(Direction.FROM)) + .getConnectorConfigurable(job.getConnectorId(Direction.FROM)) .getFromConfig(); MToConfig toConfig = ConnectorManager.getInstance() - .getConnectorConfig(job.getConnectorId(Direction.TO)) + .getConnectorConfigurable(job.getConnectorId(Direction.TO)) .getToConfig(); MDriverConfig driverConfig = Driver.getInstance().getDriver().getDriverConfig(); @@ -179,8 +179,8 @@ public class JobRequestHandler implements RequestHandler { } // Corresponding connectors for this - SqoopConnector fromConnector = ConnectorManager.getInstance().getConnector(job.getConnectorId(Direction.FROM)); - SqoopConnector toConnector = ConnectorManager.getInstance().getConnector(job.getConnectorId(Direction.TO)); + SqoopConnector fromConnector = ConnectorManager.getInstance().getSqoopConnector(job.getConnectorId(Direction.FROM)); + SqoopConnector toConnector = ConnectorManager.getInstance().getSqoopConnector(job.getConnectorId(Direction.TO)); if (!fromConnector.getSupportedDirections().contains(Direction.FROM)) { throw new SqoopException(ServerError.SERVER_0004, "Connector " + fromConnector.getClass().getCanonicalName() @@ -196,7 +196,7 @@ public class JobRequestHandler implements RequestHandler { Object fromConfigObject = ClassUtils.instantiate(fromConnector.getJobConfigurationClass(Direction.FROM)); Object toConfigObject = ClassUtils.instantiate(toConnector.getJobConfigurationClass(Direction.TO)); - Object driverConfigObject = ClassUtils.instantiate(Driver.getInstance().getDriverConfigurationGroupClass()); + Object driverConfigObject = ClassUtils.instantiate(Driver.getInstance().getDriverJobConfigurationClass()); ConfigUtils.fromConfigs(job.getJobConfig(Direction.FROM).getConfigs(), fromConfigObject); ConfigUtils.fromConfigs(job.getJobConfig(Direction.TO).getConfigs(), toConfigObject); http://git-wip-us.apache.org/repos/asf/sqoop/blob/39a22000/server/src/main/java/org/apache/sqoop/handler/LinkRequestHandler.java ---------------------------------------------------------------------- diff --git a/server/src/main/java/org/apache/sqoop/handler/LinkRequestHandler.java b/server/src/main/java/org/apache/sqoop/handler/LinkRequestHandler.java index 80e65b8..b715ad3 100644 --- a/server/src/main/java/org/apache/sqoop/handler/LinkRequestHandler.java +++ b/server/src/main/java/org/apache/sqoop/handler/LinkRequestHandler.java @@ -158,7 +158,7 @@ public class LinkRequestHandler implements RequestHandler { // Verify that user is not trying to spoof us MLinkConfig linkConfig = - ConnectorManager.getInstance().getConnectorConfig(link.getConnectorId()) + ConnectorManager.getInstance().getConnectorConfigurable(link.getConnectorId()) .getLinkConfig(); if(!linkConfig.equals(link.getConnectorLinkConfig())) { throw new SqoopException(ServerError.SERVER_0003, @@ -166,7 +166,7 @@ public class LinkRequestHandler implements RequestHandler { } // Responsible connector for this session - SqoopConnector connector = ConnectorManager.getInstance().getConnector(link.getConnectorId()); + SqoopConnector connector = ConnectorManager.getInstance().getSqoopConnector(link.getConnectorId()); // We need translate configs Object connectorLinkConfig = ClassUtils.instantiate(connector.getLinkConfigurationClass()); http://git-wip-us.apache.org/repos/asf/sqoop/blob/39a22000/spi/src/main/java/org/apache/sqoop/configurable/ConfigurableUpgradeUtil.java ---------------------------------------------------------------------- diff --git a/spi/src/main/java/org/apache/sqoop/configurable/ConfigurableUpgradeUtil.java b/spi/src/main/java/org/apache/sqoop/configurable/ConfigurableUpgradeUtil.java new file mode 100644 index 0000000..715a61c --- /dev/null +++ b/spi/src/main/java/org/apache/sqoop/configurable/ConfigurableUpgradeUtil.java @@ -0,0 +1,62 @@ +/** + * 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.configurable; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.apache.log4j.Logger; +import org.apache.sqoop.common.SqoopException; +import org.apache.sqoop.model.MConfig; +import org.apache.sqoop.model.MInput; + +public class ConfigurableUpgradeUtil { + + private static final Logger LOG = Logger.getLogger(ConfigurableUpgradeUtil.class); + /* + * For now, there is no real upgrade. So copy all data over, + * set the validation messages and error messages to be the same as for the + * inputs in the original one. + */ + @SuppressWarnings("unchecked") + public static void doUpgrade(List<MConfig> original, List<MConfig> target) { + Map<String, MConfig> configMap = new HashMap<String, MConfig>(); + for (MConfig config : original) { + configMap.put(config.getName(), config); + } + for (MConfig config : target) { + List<MInput<?>> inputs = config.getInputs(); + MConfig originalConfig = configMap.get(config.getName()); + if (originalConfig == null) { + LOG.warn("Config: '" + config.getName() + "' not present in old " + + "configurable. So it and its inputs will not be transferred by the upgrader."); + continue; + } + for (MInput input : inputs) { + try { + MInput originalInput = originalConfig.getInput(input.getName()); + input.setValue(originalInput.getValue()); + } catch (SqoopException ex) { + LOG.warn("Input: '" + input.getName() + "' not present in old " + + "configurable. So it will not be transferred by the upgrader."); + } + } + } + } +} http://git-wip-us.apache.org/repos/asf/sqoop/blob/39a22000/spi/src/main/java/org/apache/sqoop/connector/ConfigurableError.java ---------------------------------------------------------------------- diff --git a/spi/src/main/java/org/apache/sqoop/connector/ConfigurableError.java b/spi/src/main/java/org/apache/sqoop/connector/ConfigurableError.java new file mode 100644 index 0000000..4d34691 --- /dev/null +++ b/spi/src/main/java/org/apache/sqoop/connector/ConfigurableError.java @@ -0,0 +1,43 @@ +/** + * 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; + +import org.apache.sqoop.common.ErrorCode; + +public enum ConfigurableError implements ErrorCode { + + /** An unknown error has occurred. */ + CONFIGURABLE_0001("Link object upgrade called, but no upgrade routine provided for LINK config"), + CONFIGURABLE_0002("Job object upgrade called, but no upgrade routine provided for FROM job config"), + CONFIGURABLE_0003("Job object upgrade called, but no upgrade routine provided for TO job config"), + ; + private final String message; + + private ConfigurableError(String message) { + this.message = message; + } + + public String getCode() { + return name(); + } + + public String getMessage() { + return message; + } +} + http://git-wip-us.apache.org/repos/asf/sqoop/blob/39a22000/spi/src/main/java/org/apache/sqoop/connector/spi/ConnectorConfigurableUpgrader.java ---------------------------------------------------------------------- diff --git a/spi/src/main/java/org/apache/sqoop/connector/spi/ConnectorConfigurableUpgrader.java b/spi/src/main/java/org/apache/sqoop/connector/spi/ConnectorConfigurableUpgrader.java new file mode 100644 index 0000000..a112309 --- /dev/null +++ b/spi/src/main/java/org/apache/sqoop/connector/spi/ConnectorConfigurableUpgrader.java @@ -0,0 +1,84 @@ +/* + * 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.spi; + +import org.apache.sqoop.common.SqoopException; +import org.apache.sqoop.connector.ConfigurableError; +import org.apache.sqoop.model.MFromConfig; +import org.apache.sqoop.model.MLinkConfig; +import org.apache.sqoop.model.MToConfig; + +/** + * Configurable represents an entity that can provide configurations for the + * support config types {@linkplain ConfigType} + * This api represents the interface that configurable such as the connector/driver + * will implement to upgrade both the config and its corresponding data across different + * versions + * + */ +public abstract class ConnectorConfigurableUpgrader { + + /** + * Upgrade the original link config for the given config type and fill into the upgradeTarget. Note + * that any data already in {@code upgradeTarget} maybe overwritten. + * @param original - original config as in the repository + * @param upgradeTarget - the instance that will be filled in with the + * upgraded config + */ + public void upgradeLinkConfig(MLinkConfig original, MLinkConfig upgradeTarget) { + // The reasoning for throwing an exception by default is as follows. + // Sqoop calls the upgrade apis for every connector if and only if the + // corresponding link object that the config is associated with exists in the sqoop + // repository. In unexpected scenarios, if a link object is created in the + // sqoop repository without a corresponding upgrade routine for + // the link config, then this exception will be thrown to indicate a + // unexpected code path. In normal circumstances this + // scenario of having a link object for a connector without link config is + // very unlikely to happen. A likely scenario is that a connector will not have a link config and hence + // no link object will be created and thus this method will not be invoked. + throw new SqoopException(ConfigurableError.CONFIGURABLE_0001); + + } + + /** + * Upgrade the original FROM job config for the given config type and fill into the upgradeTarget. Note + * that any data already in {@code upgradeTarget} maybe overwritten. + * @param original - original config as in the repository + * @param upgradeTarget - the instance that will be filled in with the + * upgraded config + */ + + public void upgradeFromJobConfig(MFromConfig original, MFromConfig upgradeTarget) { + // see above for the reasoning behind the exception + throw new SqoopException(ConfigurableError.CONFIGURABLE_0002); + } + /** + * Upgrade the original TO job config for the given config type and fill into the upgradeTarget. Note + * that any data already in {@code upgradeTarget} maybe overwritten. + * @param original - original config as in the repository + * @param upgradeTarget - the instance that will be filled in with the + * upgraded config + */ + public void upgradeToJobConfig(MToConfig original, MToConfig upgradeTarget) { + // see above for the reasoning behind the exception + throw new SqoopException(ConfigurableError.CONFIGURABLE_0003); + + } + +} http://git-wip-us.apache.org/repos/asf/sqoop/blob/39a22000/spi/src/main/java/org/apache/sqoop/connector/spi/RepositoryUpgrader.java ---------------------------------------------------------------------- diff --git a/spi/src/main/java/org/apache/sqoop/connector/spi/RepositoryUpgrader.java b/spi/src/main/java/org/apache/sqoop/connector/spi/RepositoryUpgrader.java deleted file mode 100644 index 879e428..0000000 --- a/spi/src/main/java/org/apache/sqoop/connector/spi/RepositoryUpgrader.java +++ /dev/null @@ -1,51 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.sqoop.connector.spi; - -import org.apache.sqoop.model.MConfigList; -import org.apache.sqoop.model.MLinkConfig; - -/** - * Repository represents the sqoop entity store. Sqoop entities include - * the connectors, links, jobs and submissions and corresponding configs. - * - */ -public abstract class RepositoryUpgrader { - - /** - * Upgrade the original link config and fill into the upgradeTarget. Note - * that any data already in {@code upgradeTarget} maybe overwritten. - * @param original - original link config as in the repository - * @param upgradeTarget - the instance that will be filled in with the - * upgraded link config. - */ - public abstract void upgrade(MLinkConfig original, MLinkConfig upgradeTarget); - /** - * Upgrade the original job config and fill into the upgradeTarget. Note - * that any config data already in {@code upgradeTarget} maybe overwritten. - * This method must be called only after the link config has - * already been upgraded. - * @param original - original job config as in the repository - * @param upgradeTarget - the instance that will be filled in with the - * upgraded job config. - * NOTE(VB): This api will be revisited to accomodate from and to job config update - */ - public abstract void upgrade(MConfigList original, MConfigList upgradeTarget); -} - http://git-wip-us.apache.org/repos/asf/sqoop/blob/39a22000/spi/src/main/java/org/apache/sqoop/connector/spi/SqoopConnector.java ---------------------------------------------------------------------- diff --git a/spi/src/main/java/org/apache/sqoop/connector/spi/SqoopConnector.java b/spi/src/main/java/org/apache/sqoop/connector/spi/SqoopConnector.java index 5315e1f..6ca6c18 100644 --- a/spi/src/main/java/org/apache/sqoop/connector/spi/SqoopConnector.java +++ b/spi/src/main/java/org/apache/sqoop/connector/spi/SqoopConnector.java @@ -60,11 +60,13 @@ public abstract class SqoopConnector { /** * @return Get link configuration group class */ + @SuppressWarnings("rawtypes") public abstract Class getLinkConfigurationClass(); /** * @return Get job configuration group class per direction type or null if not supported */ + @SuppressWarnings("rawtypes") public abstract Class getJobConfigurationClass(Direction direction); /** @@ -85,11 +87,11 @@ public abstract class SqoopConnector { public abstract Validator getConfigValidator(); /** - * Returns an {@linkplain RepositoryUpgrader} object that can upgrade the + * Returns an {@linkplain ConnectorConfigurableUpgrader} object that can upgrade the * configs related to the link and job * @return RespositoryUpgrader object */ - public abstract RepositoryUpgrader getRepositoryUpgrader(); + public abstract ConnectorConfigurableUpgrader getConfigurableUpgrader(); /** * Returns the {@linkplain IntermediateDataFormat} this connector http://git-wip-us.apache.org/repos/asf/sqoop/blob/39a22000/tools/src/main/java/org/apache/sqoop/tools/tool/RepositoryDumpTool.java ---------------------------------------------------------------------- diff --git a/tools/src/main/java/org/apache/sqoop/tools/tool/RepositoryDumpTool.java b/tools/src/main/java/org/apache/sqoop/tools/tool/RepositoryDumpTool.java index f89c546..819cf6a 100644 --- a/tools/src/main/java/org/apache/sqoop/tools/tool/RepositoryDumpTool.java +++ b/tools/src/main/java/org/apache/sqoop/tools/tool/RepositoryDumpTool.java @@ -139,7 +139,7 @@ public class RepositoryDumpTool extends ConfiguredTool { while (iterator.hasNext()) { JSONObject result = iterator.next(); Long connectorId = (Long) result.get(JSONConstants.CONNECTOR_ID); - result.put(JSONConstants.CONNECTOR_NAME, connectorManager.getConnectorConfig(connectorId).getUniqueName()); + result.put(JSONConstants.CONNECTOR_NAME, connectorManager.getConnectorConfigurable(connectorId).getUniqueName()); } return json; http://git-wip-us.apache.org/repos/asf/sqoop/blob/39a22000/tools/src/main/java/org/apache/sqoop/tools/tool/RepositoryLoadTool.java ---------------------------------------------------------------------- diff --git a/tools/src/main/java/org/apache/sqoop/tools/tool/RepositoryLoadTool.java b/tools/src/main/java/org/apache/sqoop/tools/tool/RepositoryLoadTool.java index 76ebd3b..8cf9cf1 100644 --- a/tools/src/main/java/org/apache/sqoop/tools/tool/RepositoryLoadTool.java +++ b/tools/src/main/java/org/apache/sqoop/tools/tool/RepositoryLoadTool.java @@ -18,6 +18,14 @@ package org.apache.sqoop.tools.tool; +import java.io.FileInputStream; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.InputStream; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + import org.apache.commons.cli.CommandLine; import org.apache.commons.cli.CommandLineParser; import org.apache.commons.cli.GnuParser; @@ -25,13 +33,15 @@ import org.apache.commons.cli.OptionBuilder; import org.apache.commons.cli.Options; import org.apache.commons.cli.ParseException; import org.apache.commons.io.Charsets; +import org.apache.commons.io.IOUtils; import org.apache.log4j.Logger; import org.apache.sqoop.common.Direction; import org.apache.sqoop.common.VersionInfo; import org.apache.sqoop.connector.ConnectorManager; -import org.apache.sqoop.connector.spi.RepositoryUpgrader; +import org.apache.sqoop.connector.spi.ConnectorConfigurableUpgrader; import org.apache.sqoop.connector.spi.SqoopConnector; import org.apache.sqoop.driver.Driver; +import org.apache.sqoop.driver.DriverUpgrader; import org.apache.sqoop.json.JobBean; import org.apache.sqoop.json.LinkBean; import org.apache.sqoop.json.SubmissionBean; @@ -50,16 +60,6 @@ import org.apache.sqoop.model.MToConfig; import org.apache.sqoop.repository.Repository; import org.apache.sqoop.repository.RepositoryManager; import org.apache.sqoop.tools.ConfiguredTool; - -import java.io.FileInputStream; -import java.io.FileNotFoundException; -import java.io.IOException; -import java.io.InputStream; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -import org.apache.commons.io.IOUtils; import org.apache.sqoop.utils.ClassUtils; import org.apache.sqoop.validation.ConfigValidationResult; import org.apache.sqoop.validation.ConfigValidationRunner; @@ -140,8 +140,7 @@ public class RepositoryLoadTool extends ConfiguredTool { Repository repository = RepositoryManager.getInstance().getRepository(); ConnectorManager.getInstance().initialize(); - ConnectorManager connectorManager = ConnectorManager.getInstance(); - + LOG.info("Loading Connections"); JSONObject jsonConns = (JSONObject) repo.get(JSONConstants.LINKS); @@ -247,20 +246,21 @@ public class RepositoryLoadTool extends ConfiguredTool { //starting by pretending we have a brand new connection resetPersistenceId(link); - RepositoryUpgrader upgrader = Driver.getInstance().getDriverConfigRepositoryUpgrader(); Repository repository = RepositoryManager.getInstance().getRepository(); - MConnector mConnector = ConnectorManager.getInstance().getConnectorConfig(link.getConnectorId()); + MConnector mConnector = ConnectorManager.getInstance().getConnectorConfigurable(link.getConnectorId()); + ConnectorConfigurableUpgrader connectorConfigUpgrader = ConnectorManager.getInstance().getSqoopConnector(mConnector.getUniqueName()).getConfigurableUpgrader(); + List<MConfig> connectorConfigs = mConnector.getLinkConfig().clone(false).getConfigs(); MLinkConfig newLinkConfigs = new MLinkConfig(connectorConfigs); - // upgrading the forms to make sure they match the current repository - upgrader.upgrade(link.getConnectorLinkConfig(), newLinkConfigs); + // upgrading the configs to make sure they match the current repository + connectorConfigUpgrader.upgradeLinkConfig(link.getConnectorLinkConfig(), newLinkConfigs); MLink newLink = new MLink(link, newLinkConfigs); // Transform config structures to objects for validations - SqoopConnector connector = - ConnectorManager.getInstance().getConnector(link.getConnectorId()); + SqoopConnector connector = ConnectorManager.getInstance().getSqoopConnector( + link.getConnectorId()); Object connectorConfig = ClassUtils.instantiate( connector.getLinkConfigurationClass()); @@ -286,27 +286,32 @@ public class RepositoryLoadTool extends ConfiguredTool { private long loadJob(MJob job) { //starting by pretending we have a brand new job resetPersistenceId(job); + MConnector mFromConnector = ConnectorManager.getInstance().getConnectorConfigurable(job.getFromConnectorId()); + MConnector mToConnector = ConnectorManager.getInstance().getConnectorConfigurable(job.getToConnectorId()); - RepositoryUpgrader upgrader = Driver.getInstance().getDriverConfigRepositoryUpgrader(); - MDriver driver = Driver.getInstance().getDriver(); - Repository repository = RepositoryManager.getInstance().getRepository(); + MFromConfig fromConfig = job.getFromJobConfig(); + MToConfig toConfig = job.getToJobConfig(); + + ConnectorConfigurableUpgrader fromConnectorConfigUpgrader = ConnectorManager.getInstance().getSqoopConnector(mFromConnector.getUniqueName()).getConfigurableUpgrader(); + ConnectorConfigurableUpgrader toConnectorConfigUpgrader = ConnectorManager.getInstance().getSqoopConnector(mToConnector.getUniqueName()).getConfigurableUpgrader(); + + fromConnectorConfigUpgrader.upgradeFromJobConfig(job.getFromJobConfig(), fromConfig); + toConnectorConfigUpgrader.upgradeToJobConfig(job.getToJobConfig(), toConfig); + + DriverUpgrader driverConfigUpgrader = Driver.getInstance().getConfigurableUpgrader(); + MDriver driver = Driver.getInstance().getDriver(); MDriverConfig driverConfigs = driver.getDriverConfig(); - MFromConfig fromConfigs = job.getFromJobConfig(); - MToConfig toConfigs = job.getToJobConfig(); + driverConfigUpgrader.upgradeJobConfig( job.getDriverConfig(), driverConfigs); - // upgrading the configs to make sure they match the current repository - upgrader.upgrade(job.getDriverConfig(), driverConfigs); - upgrader.upgrade(job.getFromJobConfig(), fromConfigs); - upgrader.upgrade(job.getToJobConfig(), toConfigs); - MJob newJob = new MJob(job, fromConfigs, toConfigs, driverConfigs); + MJob newJob = new MJob(job, fromConfig, toConfig, driverConfigs); // Transform config structures to objects for validations SqoopConnector fromConnector = - ConnectorManager.getInstance().getConnector( + ConnectorManager.getInstance().getSqoopConnector( job.getConnectorId(Direction.FROM)); SqoopConnector toConnector = - ConnectorManager.getInstance().getConnector( + ConnectorManager.getInstance().getSqoopConnector( job.getConnectorId(Direction.TO)); Object fromConnectorConfig = ClassUtils.instantiate( @@ -314,7 +319,7 @@ public class RepositoryLoadTool extends ConfiguredTool { Object toConnectorConfig = ClassUtils.instantiate( toConnector.getJobConfigurationClass(Direction.TO)); Object driverConfig = ClassUtils.instantiate( - Driver.getInstance().getDriverConfigurationGroupClass()); + Driver.getInstance().getDriverJobConfigurationClass()); ConfigUtils.fromConfigs( job.getFromJobConfig().getConfigs(), fromConnectorConfig); @@ -332,7 +337,7 @@ public class RepositoryLoadTool extends ConfiguredTool { toConnectorConfigResult.getStatus(), driverConfigResult.getStatus()); if (finalStatus.canProceed()) { - repository.createJob(newJob); + RepositoryManager.getInstance().getRepository().createJob(newJob); } else { LOG.error("Failed to load job:" + job.getName());
