@Test public void testRoundRobinBackoffIncreasingBackoffs() throws EventDeliveryException, InterruptedException { Channel ch = new MockChannel(); int n = 100; int numEvents = 3*n; for (int i = 0; i < numEvents; i++) { ch.put(new MockEvent("test" + i)); } MockSink s1 = new MockSink(1); s1.setChannel(ch); MockSink s2 = new MockSink(2); s2.setChannel(ch); s2.setFail(true); MockSink s3 = new MockSink(3); s3.setChannel(ch); List<Sink> sinks = new ArrayList<Sink>(); sinks.add(s1); sinks.add(s2); sinks.add(s3); LoadBalancingSinkProcessor lbsp = getProcessor("round_robin",sinks, true); Status s = Status.READY; for (int i = 0; i < 3 && s != Status.BACKOFF; i++) { s = lbsp.process(); } Assert.assertEquals(0, s2.getEvents().size()); Thread.sleep(2100); for (int i = 0; i < 3 && s != Status.BACKOFF; i++) { s = lbsp.process(); } Assert.assertEquals(0, s2.getEvents().size()); s2.setFail(false); Thread.sleep(2100); for (int i = 0; i < 3 && s != Status.BACKOFF; i++) { s = lbsp.process(); } Assert.assertEquals(0, s2.getEvents().size()); Thread.sleep(2100); while (s != Status.BACKOFF) { s = lbsp.process(); } Assert.assertEquals( n + 2, s1.getEvents().size()); Assert.assertEquals( n - 3, s2.getEvents().size()); Assert.assertEquals( n + 1, s3.getEvents().size()); }
@Override public Status process() throws EventDeliveryException { Status status = null; Iterator<Sink> sinkIterator = selector.createSinkIterator(); while (sinkIterator.hasNext()) { Sink sink = sinkIterator.next(); try { status = sink.process(); break; } catch (Exception ex) { selector.informSinkFailed(sink); LOGGER.warn("Sink failed to consume event. " + "Attempting next sink if available.", ex); } } if (status == null) { throw new EventDeliveryException("All configured sinks have failed"); } return status; }
LoadBalancingSinkProcessor extends AbstractSinkProcessor { @Override public Status process() throws EventDeliveryException { Status status = null; Iterator<Sink> sinkIterator = selector.createSinkIterator(); while (sinkIterator.hasNext()) { Sink sink = sinkIterator.next(); try { status = sink.process(); break; } catch (Exception ex) { selector.informSinkFailed(sink); LOGGER.warn("Sink failed to consume event. " + "Attempting next sink if available.", ex); } } if (status == null) { throw new EventDeliveryException("All configured sinks have failed"); } return status; } }
LoadBalancingSinkProcessor extends AbstractSinkProcessor { @Override public Status process() throws EventDeliveryException { Status status = null; Iterator<Sink> sinkIterator = selector.createSinkIterator(); while (sinkIterator.hasNext()) { Sink sink = sinkIterator.next(); try { status = sink.process(); break; } catch (Exception ex) { selector.informSinkFailed(sink); LOGGER.warn("Sink failed to consume event. " + "Attempting next sink if available.", ex); } } if (status == null) { throw new EventDeliveryException("All configured sinks have failed"); } return status; }  }
LoadBalancingSinkProcessor extends AbstractSinkProcessor { @Override public Status process() throws EventDeliveryException { Status status = null; Iterator<Sink> sinkIterator = selector.createSinkIterator(); while (sinkIterator.hasNext()) { Sink sink = sinkIterator.next(); try { status = sink.process(); break; } catch (Exception ex) { selector.informSinkFailed(sink); LOGGER.warn("Sink failed to consume event. " + "Attempting next sink if available.", ex); } } if (status == null) { throw new EventDeliveryException("All configured sinks have failed"); } return status; }  @Override void configure(Context context); @Override void start(); @Override void stop(); @Override Status process(); }
LoadBalancingSinkProcessor extends AbstractSinkProcessor { @Override public Status process() throws EventDeliveryException { Status status = null; Iterator<Sink> sinkIterator = selector.createSinkIterator(); while (sinkIterator.hasNext()) { Sink sink = sinkIterator.next(); try { status = sink.process(); break; } catch (Exception ex) { selector.informSinkFailed(sink); LOGGER.warn("Sink failed to consume event. " + "Attempting next sink if available.", ex); } } if (status == null) { throw new EventDeliveryException("All configured sinks have failed"); } return status; }  @Override void configure(Context context); @Override void start(); @Override void stop(); @Override Status process(); static final String CONFIG_SELECTOR; static final String CONFIG_SELECTOR_PREFIX; static final String CONFIG_BACKOFF; static final String SELECTOR_NAME_ROUND_ROBIN; static final String SELECTOR_NAME_RANDOM; static final String SELECTOR_NAME_ROUND_ROBIN_BACKOFF; static final String SELECTOR_NAME_RANDOM_BACKOFF; }