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; }
@Test public void testDefaultConfiguration() throws Exception { 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); 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(sinks, new Context()); Status s = Status.READY; while (s != Status.BACKOFF) { s = lbsp.process(); } Assert.assertTrue(s1.getEvents().size() == n); Assert.assertTrue(s2.getEvents().size() == n); Assert.assertTrue(s3.getEvents().size() == n); }
@Test public void testRandomOneActiveSink() throws Exception { Channel ch = new MockChannel(); int n = 10; 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, false); Sink.Status s = Sink.Status.READY; while (s != Sink.Status.BACKOFF) { s = lbsp.process(); } Assert.assertTrue(s1.getEvents().size() == 0); Assert.assertTrue(s2.getEvents().size() == n); Assert.assertTrue(s3.getEvents().size() == 0); }
@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()); }
@Test public void testRandomPersistentFailure() throws Exception { 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("random",sinks, false); Status s = Status.READY; while (s != Status.BACKOFF) { s = lbsp.process(); } Assert.assertTrue(s2.getEvents().size() == 0); Assert.assertTrue(s1.getEvents().size() + s3.getEvents().size() == 3 * n); }
@Test public void testRandomNoFailure() throws Exception { Channel ch = new MockChannel(); int n = 10000; int numEvents = 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); MockSink s3 = new MockSink(3); s3.setChannel(ch); MockSink s4 = new MockSink(4); s4.setChannel(ch); MockSink s5 = new MockSink(5); s5.setChannel(ch); MockSink s6 = new MockSink(6); s6.setChannel(ch); MockSink s7 = new MockSink(7); s7.setChannel(ch); MockSink s8 = new MockSink(8); s8.setChannel(ch); MockSink s9 = new MockSink(9); s9.setChannel(ch); MockSink s0 = new MockSink(0); s0.setChannel(ch); List<Sink> sinks = new ArrayList<Sink>(); sinks.add(s1); sinks.add(s2); sinks.add(s3); sinks.add(s4); sinks.add(s5); sinks.add(s6); sinks.add(s7); sinks.add(s8); sinks.add(s9); sinks.add(s0); LoadBalancingSinkProcessor lbsp = getProcessor("random",sinks, false); Status s = Status.READY; while (s != Status.BACKOFF) { s = lbsp.process(); } Set<Integer> sizeSet = new HashSet<Integer>(); int sum = 0; for (Sink ms : sinks) { int count = ((MockSink) ms).getEvents().size(); sum += count; sizeSet.add(count); } Assert.assertEquals(n, sum); Assert.assertTrue("Miraculous distribution", sizeSet.size() > 1); }
@Test public void testRoundRobinOneActiveSink() throws Exception { Channel ch = new MockChannel(); int n = 10; 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("round_robin", sinks, false); Sink.Status s = Sink.Status.READY; while (s != Sink.Status.BACKOFF) { s = lbsp.process(); } Assert.assertTrue(s1.getEvents().size() == 0); Assert.assertTrue(s2.getEvents().size() == n); Assert.assertTrue(s3.getEvents().size() == 0); }
@Test public void testRoundRobinPersistentFailure() throws Exception { 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, false); Status s = Status.READY; while (s != Status.BACKOFF) { s = lbsp.process(); } Assert.assertTrue(s1.getEvents().size() == n); Assert.assertTrue(s2.getEvents().size() == 0); Assert.assertTrue(s3.getEvents().size() == 2 * n); }
@Test public void testRoundRobinBackoffInitialFailure() throws EventDeliveryException { 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); 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(); } s2.setFail(true); for (int i = 0; i < 3 && s != Status.BACKOFF; i++) { s = lbsp.process(); } s2.setFail(false); while (s != Status.BACKOFF) { s = lbsp.process(); } Assert.assertEquals((3 * n) / 2, s1.getEvents().size()); Assert.assertEquals(1, s2.getEvents().size()); Assert.assertEquals((3 * n) / 2 - 1, s3.getEvents().size()); }
@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()); }
@Test public void testRoundRobinBackoffFailureRecovery() 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(); } s2.setFail(false); Thread.sleep(2001); while (s != Status.BACKOFF) { s = lbsp.process(); } Assert.assertEquals(n + 1, s1.getEvents().size()); Assert.assertEquals(n - 1, s2.getEvents().size()); Assert.assertEquals(n, s3.getEvents().size()); }
@Test public void testRoundRobinNoFailure() throws Exception { 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); 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, false); Status s = Status.READY; while (s != Status.BACKOFF) { s = lbsp.process(); } Assert.assertTrue(s1.getEvents().size() == n); Assert.assertTrue(s2.getEvents().size() == n); Assert.assertTrue(s3.getEvents().size() == n); }
@Test public void testCustomSelector() throws Exception { Channel ch = new MockChannel(); int n = 10; 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); List<Sink> sinks = new ArrayList<Sink>(); sinks.add(s1); sinks.add(s2); sinks.add(s3); Context ctx = getContext(FixedOrderSelector.class.getCanonicalName()); ctx.put("selector." + FixedOrderSelector.SET_ME, "foo"); LoadBalancingSinkProcessor lbsp = getProcessor(sinks, ctx); Sink.Status s = Sink.Status.READY; while (s != Sink.Status.BACKOFF) { s = lbsp.process(); } Assert.assertTrue(s1.getEvents().size() == 0); Assert.assertTrue(s2.getEvents().size() == n); Assert.assertTrue(s3.getEvents().size() == 0); }