@Test public void testFullBatchStatus() throws Exception { when(channel.take()).thenAnswer(new Answer<Event>() { @Override public Event answer(final InvocationOnMock invocation) throws Throwable { if (counter < 10) { counter++; return EventBuilder.withBody(Integer.toString(counter).getBytes()); } return null; } }); assertSame(Status.READY, sink.process()); }
@Override public Status process() throws EventDeliveryException { final Channel channel = getChannel(); final Transaction transaction = channel.getTransaction(); try { transaction.begin(); connectionManager.ensureConnectionValid(); prepareJDBC(); int count; for (count = 0; count < batchSize; count++) { final Event event = channel.take(); if (event == null) { break; } processJDBC(event); } completeJDBC(); final Status status = updateAttemptCounters(count); connectionManager.getConnection().commit(); transaction.commit(); updateSuccessCounters(count); return status; } catch (Exception e) { try { abortJDBC(); connectionManager.getConnection().rollback(); transaction.rollback(); updateFailureCounters(); } catch (Exception e2) { LOG.error( "Exception in rollback. Rollback might not have been successful.", e2); } LOG.error("Failed to commit transaction. Transaction rolled back.", e); Throwables.propagate(e); } finally { transaction.close(); } return null; }
AbstractJDBCSink extends AbstractSink implements
    Configurable { @Override public Status process() throws EventDeliveryException { final Channel channel = getChannel(); final Transaction transaction = channel.getTransaction(); try { transaction.begin(); connectionManager.ensureConnectionValid(); prepareJDBC(); int count; for (count = 0; count < batchSize; count++) { final Event event = channel.take(); if (event == null) { break; } processJDBC(event); } completeJDBC(); final Status status = updateAttemptCounters(count); connectionManager.getConnection().commit(); transaction.commit(); updateSuccessCounters(count); return status; } catch (Exception e) { try { abortJDBC(); connectionManager.getConnection().rollback(); transaction.rollback(); updateFailureCounters(); } catch (Exception e2) { LOG.error( "Exception in rollback. Rollback might not have been successful.", e2); } LOG.error("Failed to commit transaction. Transaction rolled back.", e); Throwables.propagate(e); } finally { transaction.close(); } return null; } }
AbstractJDBCSink extends AbstractSink implements
    Configurable { @Override public Status process() throws EventDeliveryException { final Channel channel = getChannel(); final Transaction transaction = channel.getTransaction(); try { transaction.begin(); connectionManager.ensureConnectionValid(); prepareJDBC(); int count; for (count = 0; count < batchSize; count++) { final Event event = channel.take(); if (event == null) { break; } processJDBC(event); } completeJDBC(); final Status status = updateAttemptCounters(count); connectionManager.getConnection().commit(); transaction.commit(); updateSuccessCounters(count); return status; } catch (Exception e) { try { abortJDBC(); connectionManager.getConnection().rollback(); transaction.rollback(); updateFailureCounters(); } catch (Exception e2) { LOG.error( "Exception in rollback. Rollback might not have been successful.", e2); } LOG.error("Failed to commit transaction. Transaction rolled back.", e); Throwables.propagate(e); } finally { transaction.close(); } return null; }  }
AbstractJDBCSink extends AbstractSink implements
    Configurable { @Override public Status process() throws EventDeliveryException { final Channel channel = getChannel(); final Transaction transaction = channel.getTransaction(); try { transaction.begin(); connectionManager.ensureConnectionValid(); prepareJDBC(); int count; for (count = 0; count < batchSize; count++) { final Event event = channel.take(); if (event == null) { break; } processJDBC(event); } completeJDBC(); final Status status = updateAttemptCounters(count); connectionManager.getConnection().commit(); transaction.commit(); updateSuccessCounters(count); return status; } catch (Exception e) { try { abortJDBC(); connectionManager.getConnection().rollback(); transaction.rollback(); updateFailureCounters(); } catch (Exception e2) { LOG.error( "Exception in rollback. Rollback might not have been successful.", e2); } LOG.error("Failed to commit transaction. Transaction rolled back.", e); Throwables.propagate(e); } finally { transaction.close(); } return null; }  @Override void configure(final Context context); @Override synchronized void start(); @Override synchronized void stop(); Connection getConnection(); @Override Status process(); }
AbstractJDBCSink extends AbstractSink implements
    Configurable { @Override public Status process() throws EventDeliveryException { final Channel channel = getChannel(); final Transaction transaction = channel.getTransaction(); try { transaction.begin(); connectionManager.ensureConnectionValid(); prepareJDBC(); int count; for (count = 0; count < batchSize; count++) { final Event event = channel.take(); if (event == null) { break; } processJDBC(event); } completeJDBC(); final Status status = updateAttemptCounters(count); connectionManager.getConnection().commit(); transaction.commit(); updateSuccessCounters(count); return status; } catch (Exception e) { try { abortJDBC(); connectionManager.getConnection().rollback(); transaction.rollback(); updateFailureCounters(); } catch (Exception e2) { LOG.error( "Exception in rollback. Rollback might not have been successful.", e2); } LOG.error("Failed to commit transaction. Transaction rolled back.", e); Throwables.propagate(e); } finally { transaction.close(); } return null; }  @Override void configure(final Context context); @Override synchronized void start(); @Override synchronized void stop(); Connection getConnection(); @Override Status process();  }