@Test public void testRandomBackoff() throws Exception { Channel ch = new MockChannel(); int n = 100; int numEvents = n; for (int i = 0; i < numEvents; i++) { ch.put(new MockEvent("test" + i)); } MockSink s1 = new MockSink(1); s1.setChannel(ch); s1.setFail(true); MockSink s2 = new MockSink(2); s2.setChannel(ch); MockSink s3 = new MockSink(3); s3.setChannel(ch); s3.setFail(true); List<Sink> sinks = new ArrayList<Sink>(); sinks.add(s1); sinks.add(s2); sinks.add(s3); LoadBalancingSinkProcessor lbsp = getProcessor("random", sinks, true); for(int i=0; i < 50; i++) { lbsp.process(); } Assert.assertEquals(50, s2.getEvents().size()); s2.setFail(true); s1.setFail(false); try { lbsp.process(); Assert.fail("Expected EventDeliveryException"); } catch (EventDeliveryException e) { } Thread.sleep(2100); Sink.Status s = Sink.Status.READY; while (s != Sink.Status.BACKOFF) { s = lbsp.process(); } Assert.assertEquals(50, s1.getEvents().size()); Assert.assertEquals(50, s2.getEvents().size()); Assert.assertEquals(0, 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; }