AgentFailChainSink extends EventSink.Base { public static String genE2EChain(String... chain) { String body = " %s "; String spec = FailoverChainManager.genAvailableSinkSpec(body, Arrays .asList(chain)); spec = "{ ackedWriteAhead => { stubbornAppend => { insistentOpen => " + spec + " } } }"; LOG.info("Setting e2e failover chain to " + spec); return spec; } AgentFailChainSink(Context context, RELIABILITY rel, String... hosts); @Override void open(); @Override void close(); @Override void append(Event e); @Override ReportEvent getMetrics(); @Override Map<String, Reportable> getSubMetrics(); @Deprecated @Override void getReports(String namePrefix, Map<String, ReportEvent> reports); static String genBestEffortChain(String... chain); static String genE2EChain(String... chain); static String genDfoChain(String... chain); static List<String> thriftifyArgs(int defaultPort, List<String> list); static SinkBuilder e2eBuilder(); static SinkBuilder dfoBuilder(); static SinkBuilder beBuilder(); @Override String getName();  }
@Test public void testWALFailchain() throws FlumeSpecException, JSONException { ReportManager.get().clear(); String spec = AgentFailChainSink.genE2EChain("counter(\"foo1\")", "counter(\"foo2\")", "counter(\"foo3\")"); LOG.info("waled failchain: " + spec); String node = NetUtils.localhost(); EventSink snk = FlumeBuilder.buildSink(new LogicalNodeContext(node, node), spec); ReportEvent rpt = ReportUtil.getFlattenedReport(snk); LOG.info(ReportUtil.toJSONObject(rpt).toString()); assertEquals(0, (long) rpt.getLongMetric("naiveWal.loggedQ")); assertEquals("foo1", rpt.getStringMetric("drainSink.StubbornAppend.InsistentOpen." + "FailoverChainSink.primary.foo1.name")); assertEquals("foo2", rpt.getStringMetric("drainSink.StubbornAppend.InsistentOpen." + "FailoverChainSink.backup.BackoffFailover.primary.foo2.name")); assertEquals("foo3", rpt.getStringMetric("drainSink.StubbornAppend.InsistentOpen." + "FailoverChainSink.backup.BackoffFailover.backup.foo3.name")); }