AbstractJDBCImporter extends AbstractShardingScalingExecutor<IncrementalPosition> implements Importer { private void execute(final Connection connection, final Record record) throws SQLException { if (DataRecord.class.equals(record.getClass())) { DataRecord dataRecord = (DataRecord) record; switch (dataRecord.getType()) { case ScalingConstant.INSERT: executeInsert(connection, dataRecord); break; case ScalingConstant.UPDATE: executeUpdate(connection, dataRecord); break; case ScalingConstant.DELETE: executeDelete(connection, dataRecord); break; default: break; } } } protected  AbstractJDBCImporter(final ImporterConfiguration importerConfig, final DataSourceManager dataSourceManager); @Override final void start(); @Override final void write();  }
@Test public void assertWriteInsertDataRecord() throws SQLException { DataRecord insertRecord = getDataRecord("INSERT"); when(sqlBuilder.buildInsertSQL(insertRecord)).thenReturn(INSERT_SQL); when(connection.prepareStatement(INSERT_SQL)).thenReturn(preparedStatement); when(channel.fetchRecords(100, 3)).thenReturn(mockRecords(insertRecord)); jdbcImporter.run(); verify(preparedStatement).setObject(1, 1); verify(preparedStatement).setObject(2, 10); verify(preparedStatement).setObject(3, "INSERT"); verify(preparedStatement).execute(); }
@Test public void assertDeleteDataRecord() throws SQLException { DataRecord deleteRecord = getDataRecord("DELETE"); when(sqlBuilder.buildDeleteSQL(deleteRecord, mockConditionColumns(deleteRecord))).thenReturn(DELETE_SQL); when(connection.prepareStatement(DELETE_SQL)).thenReturn(preparedStatement); when(channel.fetchRecords(100, 3)).thenReturn(mockRecords(deleteRecord)); jdbcImporter.run(); verify(preparedStatement).setObject(1, 1); verify(preparedStatement).setObject(2, 10); verify(preparedStatement).execute(); }
@Test public void assertUpdateDataRecord() throws SQLException { DataRecord updateRecord = getDataRecord("UPDATE"); when(sqlBuilder.buildUpdateSQL(updateRecord, mockConditionColumns(updateRecord))).thenReturn(UPDATE_SQL); when(connection.prepareStatement(UPDATE_SQL)).thenReturn(preparedStatement); when(channel.fetchRecords(100, 3)).thenReturn(mockRecords(updateRecord)); jdbcImporter.run(); verify(preparedStatement).setObject(1, 10); verify(preparedStatement).setObject(2, "UPDATE"); verify(preparedStatement).setObject(3, 1); verify(preparedStatement).setObject(4, 10); verify(preparedStatement).execute(); }