Repository: sqoop Updated Branches: refs/heads/trunk c339b23b6 -> 7808c6bc3
SQOOP-2971: OraOop does not close connections properly (Szabolcs Vasas via Kate Ting) Project: http://git-wip-us.apache.org/repos/asf/sqoop/repo Commit: http://git-wip-us.apache.org/repos/asf/sqoop/commit/7808c6bc Tree: http://git-wip-us.apache.org/repos/asf/sqoop/tree/7808c6bc Diff: http://git-wip-us.apache.org/repos/asf/sqoop/diff/7808c6bc Branch: refs/heads/trunk Commit: 7808c6bc3e7116534fb2bf83472c8fa8d2d8a0f4 Parents: c339b23 Author: Kate Ting <[email protected]> Authored: Fri Jul 1 16:00:03 2016 -0700 Committer: Kate Ting <[email protected]> Committed: Fri Jul 1 16:00:03 2016 -0700 ---------------------------------------------------------------------- .../oracle/OraOopDataDrivenDBInputFormat.java | 14 ++- .../sqoop/mapreduce/db/DBInputFormat.java | 9 +- .../com/cloudera/sqoop/ThirdPartyTests.java | 2 + ...aDrivenDBInputFormatConnectionCloseTest.java | 91 ++++++++++++++++++++ 4 files changed, 110 insertions(+), 6 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/sqoop/blob/7808c6bc/src/java/org/apache/sqoop/manager/oracle/OraOopDataDrivenDBInputFormat.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/sqoop/manager/oracle/OraOopDataDrivenDBInputFormat.java b/src/java/org/apache/sqoop/manager/oracle/OraOopDataDrivenDBInputFormat.java index 13f05d5..3e88d04 100644 --- a/src/java/org/apache/sqoop/manager/oracle/OraOopDataDrivenDBInputFormat.java +++ b/src/java/org/apache/sqoop/manager/oracle/OraOopDataDrivenDBInputFormat.java @@ -136,8 +136,16 @@ public class OraOopDataDrivenDBInputFormat<T extends SqoopRecord> extends } } + connection.commit(); } catch (SQLException ex) { + try { + connection.rollback(); + } catch (SQLException e) { + LOG.error("Cannot rollback transaction.", e); + } throw new IOException(ex); + } finally { + closeConnection(); } return splits; @@ -199,13 +207,11 @@ public class OraOopDataDrivenDBInputFormat<T extends SqoopRecord> extends // the current // value of the URL_PROPERTY... - return new OraOopDBRecordReader<T>(split, inputClass, conf, dbConf - .getConnection(), dbConf, dbConf.getInputConditions(), dbConf + return new OraOopDBRecordReader<T>(split, inputClass, conf, + getConnection(), dbConf, dbConf.getInputConditions(), dbConf .getInputFieldNames(), dbConf.getInputTableName()); } catch (SQLException ex) { throw new IOException(ex); - } catch (ClassNotFoundException ex) { - throw new IOException(ex); } } http://git-wip-us.apache.org/repos/asf/sqoop/blob/7808c6bc/src/java/org/apache/sqoop/mapreduce/db/DBInputFormat.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/sqoop/mapreduce/db/DBInputFormat.java b/src/java/org/apache/sqoop/mapreduce/db/DBInputFormat.java index 3a8e5d0..0a2e396 100644 --- a/src/java/org/apache/sqoop/mapreduce/db/DBInputFormat.java +++ b/src/java/org/apache/sqoop/mapreduce/db/DBInputFormat.java @@ -158,8 +158,11 @@ extends InputFormat<LongWritable, T> implements Configurable { @Override /** {@inheritDoc} */ public void setConf(Configuration conf) { + setDbConf(new DBConfiguration(conf)); + } - dbConf = new DBConfiguration(conf); + public void setDbConf(DBConfiguration dbConf) { + this.dbConf = dbConf; try { getConnection(); @@ -389,7 +392,9 @@ extends InputFormat<LongWritable, T> implements Configurable { this.connection.close(); this.connection = null; } - } catch (SQLException sqlE) { /* ignore exception on close. */ } + } catch (SQLException sqlE) { + LOG.error("Cannot close JDBC connection.", sqlE); + } } } http://git-wip-us.apache.org/repos/asf/sqoop/blob/7808c6bc/src/test/com/cloudera/sqoop/ThirdPartyTests.java ---------------------------------------------------------------------- diff --git a/src/test/com/cloudera/sqoop/ThirdPartyTests.java b/src/test/com/cloudera/sqoop/ThirdPartyTests.java index 50f3192..3103bd4 100644 --- a/src/test/com/cloudera/sqoop/ThirdPartyTests.java +++ b/src/test/com/cloudera/sqoop/ThirdPartyTests.java @@ -55,6 +55,7 @@ import org.apache.sqoop.manager.netezza.DirectNetezzaHCatExportManualTest; import org.apache.sqoop.manager.netezza.DirectNetezzaHCatImportManualTest; import org.apache.sqoop.manager.netezza.NetezzaExportManualTest; import org.apache.sqoop.manager.netezza.NetezzaImportManualTest; +import org.apache.sqoop.manager.oracle.OraOopDataDrivenDBInputFormatConnectionCloseTest; import org.apache.sqoop.manager.oracle.OracleCallExportTest; import org.apache.sqoop.manager.oracle.OracleIncrementalImportTest; import org.apache.sqoop.manager.oracle.OracleSplitterTest; @@ -98,6 +99,7 @@ public final class ThirdPartyTests extends TestCase { suite.addTestSuite(OracleCompatTest.class); suite.addTestSuite(OracleIncrementalImportTest.class); suite.addTestSuite(OracleSplitterTest.class); + suite.addTestSuite(OraOopDataDrivenDBInputFormatConnectionCloseTest.class); // SQL Server suite.addTestSuite(SQLServerDatatypeExportDelimitedFileManualTest.class); http://git-wip-us.apache.org/repos/asf/sqoop/blob/7808c6bc/src/test/org/apache/sqoop/manager/oracle/OraOopDataDrivenDBInputFormatConnectionCloseTest.java ---------------------------------------------------------------------- diff --git a/src/test/org/apache/sqoop/manager/oracle/OraOopDataDrivenDBInputFormatConnectionCloseTest.java b/src/test/org/apache/sqoop/manager/oracle/OraOopDataDrivenDBInputFormatConnectionCloseTest.java new file mode 100644 index 0000000..7845160 --- /dev/null +++ b/src/test/org/apache/sqoop/manager/oracle/OraOopDataDrivenDBInputFormatConnectionCloseTest.java @@ -0,0 +1,91 @@ +package org.apache.sqoop.manager.oracle; + +import com.cloudera.sqoop.mapreduce.db.DBConfiguration; +import junit.framework.TestCase; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.mapreduce.JobContext; +import org.junit.Before; +import org.junit.Test; + +import java.io.IOException; +import java.sql.Connection; +import java.sql.DatabaseMetaData; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; + +import static org.mockito.Matchers.anyString; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class OraOopDataDrivenDBInputFormatConnectionCloseTest extends TestCase { + + private static final OraOopLog LOG = OraOopLogFactory.getLog( + TestOraOopDataDrivenDBInputFormat.class.getName()); + + private static final String ORACLE_PREPARED_STATEMENT_CLASS = "oracle.jdbc.OraclePreparedStatement"; + + private OraOopDataDrivenDBInputFormat inputFormat; + + private Connection mockConnection; + + private JobContext mockJobContext; + + @Before + public void setUp() throws Exception { + Configuration configuration = new Configuration(); + configuration.set(DBConfiguration.USERNAME_PROPERTY, "Oracle user"); + configuration.setInt(OraOopConstants.ORAOOP_DESIRED_NUMBER_OF_MAPPERS, 1); + + Class<? extends PreparedStatement> preparedStatementClass = + (Class<? extends PreparedStatement>) Class.forName(ORACLE_PREPARED_STATEMENT_CLASS); + PreparedStatement mockPreparedStatement = mock(preparedStatementClass); + ResultSet mockResultSet = mock(ResultSet.class); + when(mockResultSet.next()).thenReturn(true).thenReturn(false); + when(mockPreparedStatement.executeQuery()).thenReturn(mockResultSet); + + mockConnection = mock(Connection.class); + DatabaseMetaData dbMetaData = mock(DatabaseMetaData.class); + when(dbMetaData.getDatabaseProductName()).thenReturn("Oracle"); + when(mockConnection.getMetaData()).thenReturn(dbMetaData); + when(mockConnection.prepareStatement(anyString())).thenReturn(mockPreparedStatement); + + DBConfiguration dbConf = mock(DBConfiguration.class); + when(dbConf.getConnection()).thenReturn(mockConnection); + when(dbConf.getConf()).thenReturn(configuration); + when(dbConf.getInputTableName()).thenReturn("InputTable"); + + mockJobContext = mock(JobContext.class); + when(mockJobContext.getConfiguration()).thenReturn(configuration); + + inputFormat = new OraOopDataDrivenDBInputFormat(); + inputFormat.setDbConf(dbConf); + } + + @Test + public void testGetSplitsClosesConnectionProperly() throws Exception { + inputFormat.getSplits(mockJobContext); + verify(mockConnection).commit(); + verify(mockConnection).close(); + } + + @Test + public void testGetSplitsClosesConnectionProperlyWhenExceptionIsThrown() throws Exception { + + doThrow(new SQLException("For the sake of testing the commit fails.")).when(mockConnection).commit(); + + try { + inputFormat.getSplits(mockJobContext); + } catch (IOException e) { + LOG.debug("An expected exception is thrown in testSplitsClosesConnectionProperlyWhenExceptionIsThrown, ignoring."); + } + + verify(mockConnection).rollback(); + verify(mockConnection).close(); + + } + + +}
