JDBCOutputFormat extends RichOutputFormat<Row> { void flush() { try { upload.executeBatch(); batchCount = 0; } catch (SQLException e) { throw new RuntimeException("Execution of JDBC statement failed.", e); } } JDBCOutputFormat(); @Override void configure(Configuration parameters); @Override void open(int taskNumber, int numTasks); @Override void writeRecord(Row row); @Override void close(); static JDBCOutputFormatBuilder buildJDBCOutputFormat();  }
@Test public void testFlush() throws SQLException, IOException { jdbcOutputFormat = JDBCOutputFormat.buildJDBCOutputFormat() .setDrivername(DRIVER_CLASS) .setDBUrl(DB_URL) .setQuery(String.format(INSERT_TEMPLATE, OUTPUT_TABLE_2)) .setBatchInterval(3) .finish(); try ( Connection dbConn = DriverManager.getConnection(DB_URL); PreparedStatement statement = dbConn.prepareStatement(JDBCTestBase.SELECT_ALL_NEWBOOKS_2) ) { jdbcOutputFormat.open(0, 1); for (int i = 0; i < 2; ++i) { jdbcOutputFormat.writeRecord(toRow(TEST_DATA[i])); } try (ResultSet resultSet = statement.executeQuery()) { assertFalse(resultSet.next()); } jdbcOutputFormat.writeRecord(toRow(TEST_DATA[2])); try (ResultSet resultSet = statement.executeQuery()) { int recordCount = 0; while (resultSet.next()) { assertEquals(TEST_DATA[recordCount].id, resultSet.getObject("id")); assertEquals(TEST_DATA[recordCount].title, resultSet.getObject("title")); assertEquals(TEST_DATA[recordCount].author, resultSet.getObject("author")); assertEquals(TEST_DATA[recordCount].price, resultSet.getObject("price")); assertEquals(TEST_DATA[recordCount].qty, resultSet.getObject("qty")); recordCount++; } assertEquals(3, recordCount); } } finally { jdbcOutputFormat.close(); } }