FailoverSinkProcessor extends AbstractSinkProcessor { @Override public void configure(Context context) { liveSinks = new TreeMap<Integer, Sink>(); failedSinks = new PriorityQueue<FailedSink>(); Integer nextPrio = 0; String maxPenaltyStr = context.getString(MAX_PENALTY_PREFIX); if (maxPenaltyStr == null) { maxPenalty = DEFAULT_MAX_PENALTY; } else { try { maxPenalty = Integer.parseInt(maxPenaltyStr); } catch (NumberFormatException e) { logger.warn("{} is not a valid value for {}", new Object[] { maxPenaltyStr, MAX_PENALTY_PREFIX }); maxPenalty = DEFAULT_MAX_PENALTY; } } for (Entry<String, Sink> entry : sinks.entrySet()) { String priStr = PRIORITY_PREFIX + entry.getKey(); Integer priority; try { priority = Integer.parseInt(context.getString(priStr)); } catch (Exception e) { priority = --nextPrio; } if (!liveSinks.containsKey(priority)) { liveSinks.put(priority, sinks.get(entry.getKey())); } else { logger.warn("Sink {} not added to FailverSinkProcessor as priority" + "duplicates that of sink {}", entry.getKey(), liveSinks.get(priority)); } } activeSink = liveSinks.get(liveSinks.lastKey()); }  @Override void configure(Context context); @Override Status process(); @Override void setSinks(List<Sink> sinks);  }
@Test public void testFailover() throws InterruptedException { Channel ch = new MemoryChannel(); ConsumeXSink s1 = new ConsumeXSink(10); s1.setChannel(ch); s1.setName("s1"); ConsumeXSink s2 = new ConsumeXSink(50); s2.setChannel(ch); s2.setName("s2"); ConsumeXSink s3 = new ConsumeXSink(100); s3.setChannel(ch); s3.setName("s3"); Context context = new Context(); Configurables.configure(s1, context); Configurables.configure(s2, context); Configurables.configure(s3, context); Configurables.configure(ch, context); ch.start(); List<Sink> sinks = new LinkedList<Sink>(); sinks.add(s1); sinks.add(s2); sinks.add(s3); SinkGroup group = new SinkGroup(sinks); Map<String, String> params = new HashMap<String, String>(); params.put("sinks", "s1 s2 s3"); params.put("processor.type", "failover"); params.put("processor.priority.s1", "3"); params.put("processor.priority.s2", "2"); params.put("processor.priority.s3", "1"); params.put("processor.maxpenalty", "10000"); context.putAll(params); Configurables.configure(group, context); SinkRunner runner = new SinkRunner(group.getProcessor()); runner.start(); Assert.assertEquals(LifecycleState.START, s1.getLifecycleState()); Assert.assertEquals(LifecycleState.START, s2.getLifecycleState()); Assert.assertEquals(LifecycleState.START, s3.getLifecycleState()); for (int i = 0; i < 15; i++) { Transaction tx = ch.getTransaction(); tx.begin(); ch.put(EventBuilder.withBody("test".getBytes())); tx.commit(); tx.close(); } Thread.sleep(100); Assert.assertEquals(new Integer(10), s1.getWritten()); Assert.assertEquals(new Integer(5), s2.getWritten()); for (int i = 0; i < 50; i++) { Transaction tx = ch.getTransaction(); tx.begin(); ch.put(EventBuilder.withBody("test".getBytes())); tx.commit(); tx.close(); } Thread.sleep(100); Assert.assertEquals(new Integer(50), s2.getWritten()); Assert.assertEquals(new Integer(5), s3.getWritten()); s2.setRemaining(20); Thread.sleep(5000); for (int i = 0; i < 100; i++) { Transaction tx = ch.getTransaction(); tx.begin(); ch.put(EventBuilder.withBody("test".getBytes())); tx.commit(); tx.close(); } Thread.sleep(1000); Assert.assertEquals(new Integer(10), s1.getWritten()); Assert.assertEquals(new Integer(70), s2.getWritten()); Assert.assertEquals(new Integer(85), s3.getWritten()); runner.stop(); ch.stop(); }