LogicalNode implements Reportable { public Driver getDriver() { return driver; } LogicalNode();  LogicalNode(Context ctx, String name); void loadConfig(FlumeConfigData cfg); synchronized boolean checkConfig(FlumeConfigData data); @Deprecated synchronized void getReports(Map<String, ReportEvent> reports); @Deprecated synchronized ReportEvent getReport(); ReportEvent getMetrics(); @Override Map<String, Reportable> getSubMetrics(); String getName(); long getConfigVersion(); NodeStatus getStatus(); Driver getDriver(); void close(); EventSink getSink(); EventSource getSource(); static final long VERSION_INFIMUM; static final String A_RECONFIGURES; }
@Test public void testFailedConfig() throws IOException, RuntimeException, FlumeSpecException, InterruptedException { LogicalNode node = drive("fail(\"null\")", "null"); assertEquals(DriverState.ERROR, node.getDriver().getState()); }
@Test public void testFailOpenSource() throws IOException, RuntimeException, FlumeSpecException, InterruptedException { SourceFactoryImpl srcfact = new SourceFactoryImpl(); srcfact.setSource("failOpen", new SourceBuilder() { @Override public EventSource build(Context ctx, String... argv) { return new EventSource.Base() { @Override public void open() throws IOException { LOG.info("in FailOpenSource,open"); throw new IOException("open always fails"); } }; } }); FlumeBuilder.setSourceFactory(srcfact); LogicalNode node = drive("failOpen", "null"); assertEquals(DriverState.ERROR, node.getDriver().getState()); }
@Test public void testFailNextSource() throws IOException, RuntimeException, FlumeSpecException, InterruptedException { SourceFactoryImpl srcfact = new SourceFactoryImpl(); srcfact.setSource("failNext", new SourceBuilder() { @Override public EventSource build(Context ctx, String... argv) { return new EventSource.Base() { @Override public Event next() throws IOException { LOG.info("in FailOpenSource.next"); throw new IOException("next always fails"); } }; } }); FlumeBuilder.setSourceFactory(srcfact); LogicalNode node = drive("failNext", "null"); assertEquals(DriverState.ERROR, node.getDriver().getState()); }
@Test public void testFailCloseSource() throws IOException, RuntimeException, FlumeSpecException, InterruptedException { SourceFactoryImpl srcfact = new SourceFactoryImpl(); srcfact.setSource("failClose", new SourceBuilder() { @Override public EventSource build(Context ctx, String... argv) { return new EventSource.Base() { @Override public void close() throws IOException { LOG.info("in FailOpenSource.close"); throw new IOException("close always fails"); } }; } }); FlumeBuilder.setSourceFactory(srcfact); LogicalNode node = drive("failClose", "null"); assertEquals(DriverState.ERROR, node.getDriver().getState()); }
@Test public void testFailOpenSink() throws IOException, RuntimeException, FlumeSpecException, InterruptedException { SinkFactoryImpl snkfact = new SinkFactoryImpl(); snkfact.setSink("failOpen", new SinkBuilder() { @Override public EventSink build(Context context, String... argv) { return new EventSink.Base() { @Override public void open() throws IOException { throw new IOException("open always fails"); } }; } }); FlumeBuilder.setSinkFactory(snkfact); LogicalNode node = drive("asciisynth(10)", "failOpen"); assertEquals(DriverState.ERROR, node.getDriver().getState()); }
@Test public void testFailAppendSink() throws IOException, RuntimeException, FlumeSpecException, InterruptedException { SinkFactoryImpl snkfact = new SinkFactoryImpl(); snkfact.setSink("failAppend", new SinkBuilder() { @Override public EventSink build(Context context, String... argv) { return new EventSink.Base() { @Override public void append(Event e) throws IOException { throw new IOException("sink.append always fails"); } }; } }); FlumeBuilder.setSinkFactory(snkfact); LogicalNode node = drive("asciisynth(10)", "failAppend"); assertEquals(DriverState.ERROR, node.getDriver().getState()); }
@Test public void testFailCloseSink() throws IOException, RuntimeException, FlumeSpecException, InterruptedException { SinkFactoryImpl snkfact = new SinkFactoryImpl(); snkfact.setSink("failClose", new SinkBuilder() { @Override public EventSink build(Context context, String... argv) { return new EventSink.Base() { @Override public void open() throws IOException { throw new IOException("open always fails"); } }; } }); FlumeBuilder.setSinkFactory(snkfact); LogicalNode node = drive("asciisynth(10)", "failClose"); assertEquals(DriverState.ERROR, node.getDriver().getState()); }