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) { 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 SELECTOR_NAME_ROUND_ROBIN; static final String SELECTOR_NAME_RANDOM; }
@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(); MockSink s2 = new MockSink(2); s2.setChannel(ch); MockSink s3 = new MockSink(3); s3.setChannel(ch); s3.setFail(); List<Sink> sinks = new ArrayList<Sink>(); sinks.add(s1); sinks.add(s2); sinks.add(s3); LoadBalancingSinkProcessor lbsp = getProcessor("random", sinks); 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 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(); 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); 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); 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(); MockSink s2 = new MockSink(2); s2.setChannel(ch); MockSink s3 = new MockSink(3); s3.setChannel(ch); s3.setFail(); List<Sink> sinks = new ArrayList<Sink>(); sinks.add(s1); sinks.add(s2); sinks.add(s3); LoadBalancingSinkProcessor lbsp = getProcessor("round_robin", sinks); 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(); 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); 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 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); 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(); 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); }