AgentSink extends EventSink.Base { public AgentSink(Context ctx, String dsthost, int port, ReliabilityMode mode) throws FlumeSpecException { Preconditions.checkNotNull(dsthost); String batchGzDeco = ""; String batchN = ctx.getValue(BATCH_COUNT); int n = 1; if (batchN != null) { n = Integer.parseInt(batchN); } String batchLatency = ctx.getValue(BATCH_MILLIS); int ms = 0; if (batchLatency != null) { ms = Integer.parseInt(batchLatency); } if (n > 1) { batchGzDeco += " batch(" + n + "," + ms + ") "; } if (ctx.getValue("compression") != null) { batchGzDeco += " gzip "; } switch (mode) { case ENDTOEND: { String snk = String.format("ackedWriteAhead " + batchGzDeco + " stubbornAppend insistentOpen rpcSink(\"%s\", %d)", dsthost, port); sink = FlumeBuilder.buildSink(ctx, snk); break; } case DISK_FAILOVER: { FlumeConfiguration conf = FlumeConfiguration.get(); long maxSingleBo = conf.getFailoverMaxSingleBackoff(); long initialBo = conf.getFailoverInitialBackoff(); long maxCumulativeBo = conf.getFailoverMaxCumulativeBackoff(); String rpc = String.format("rpcSink(\"%s\", %d)", dsthost, port); String snk = String.format(batchGzDeco + " < %s ? diskFailover insistentAppend " + " stubbornAppend insistentOpen(%d,%d,%d) %s >", rpc, maxSingleBo, initialBo, maxCumulativeBo, rpc); sink = FlumeBuilder.buildSink(ctx, snk); break; } case BEST_EFFORT: { String snk = String.format("< " + batchGzDeco + " insistentOpen stubbornAppend rpcSink(\"%s\", %d) ? null>", dsthost, port); sink = FlumeBuilder.buildSink(ctx, snk); break; } default: throw new FlumeSpecException("unexpected agent mode: " + mode + "!"); } } AgentSink(Context ctx, String dsthost, int port, ReliabilityMode mode); @Override void append(Event e); @Override void close(); @Override void open(); static SinkBuilder e2eBuilder(); static SinkBuilder dfoBuilder(); static SinkBuilder beBuilder(); @Override String getName(); @Override ReportEvent getMetrics(); @Override Map<String, Reportable> getSubMetrics(); @Deprecated @Override void getReports(String namePrefix, Map<String, ReportEvent> reports); static final String BATCH_COUNT; static final String BATCH_MILLIS; }
@Test public void testAgentSink() throws FlumeSpecException, IOException, InterruptedException { String snkcfg = "agentSink(\"localhost\", 12345)"; EventSource src = FlumeBuilder.buildSource( LogicalNodeContext.testingContext(), "collectorSource(12345)"); src.open(); for (int i = 0; i < 20; i++) { EventSink snk = FlumeBuilder.buildSink( LogicalNodeContext.testingContext(), snkcfg); snk.open(); snk.close(); } }