AvroSink extends AbstractSink implements Configurable { @Override public Status process() throws EventDeliveryException { Status status = Status.READY; Channel channel = getChannel(); Transaction transaction = channel.getTransaction(); try { transaction.begin(); verifyConnection(); List<Event> batch = Lists.newLinkedList(); for (int i = 0; i < batchSize; i++) { Event event = channel.take(); if (event == null) { counterGroup.incrementAndGet("batch.underflow"); break; } batch.add(event); } if (batch.isEmpty()) { counterGroup.incrementAndGet("batch.empty"); status = Status.BACKOFF; } else { client.appendBatch(batch); } transaction.commit(); counterGroup.incrementAndGet("batch.success"); } catch (ChannelException e) { transaction.rollback(); logger.error("Unable to get event from channel. Exception follows.", e); status = Status.BACKOFF; } catch (EventDeliveryException e) { transaction.rollback(); destroyConnection(); throw e; } catch (FlumeException e) { transaction.rollback(); destroyConnection(); throw new EventDeliveryException("RPC connection error. " + "Exception follows.", e); } catch (Exception e) { transaction.rollback(); destroyConnection(); throw new EventDeliveryException("Unexpected error. " + "Exception follows.", e); } finally { transaction.close(); } return status; } AvroSink(); @Override void configure(Context context); @Override void start(); @Override void stop(); @Override Status process();  }
@Test public void testProcess() throws InterruptedException, EventDeliveryException { Event event = EventBuilder.withBody("test event 1".getBytes(), new HashMap<String, String>()); Server server = createServer(); server.start(); sink.start(); Assert.assertTrue(LifecycleController.waitForOneOf(sink, LifecycleState.START_OR_ERROR, 5000)); Transaction transaction = channel.getTransaction(); transaction.begin(); for (int i = 0; i < 10; i++) { channel.put(event); } transaction.commit(); transaction.close(); for (int i = 0; i < 5; i++) { Sink.Status status = sink.process(); Assert.assertEquals(Sink.Status.READY, status); } Assert.assertEquals(Sink.Status.BACKOFF, sink.process()); sink.stop(); Assert.assertTrue(LifecycleController.waitForOneOf(sink, LifecycleState.STOP_OR_ERROR, 5000)); server.close(); }