### Question:
CollectorSource extends EventSource.Base { public CollectorSource(int port) { this.src = new ThriftEventSource(port); this.port = port; } CollectorSource(int port); @Override void close(); @Override void open(); @Override Event next(); @Override void getReports(String namePrefix, Map<String, ReportEvent> reports); int getPort(); static SourceBuilder builder();  }### Answer:
@Test public void testCollectorSource() throws FlumeSpecException, IOException { EventSource src = FlumeBuilder.buildSource("collectorSource(34568)"); EventSink snk = FlumeBuilder.buildSink(new Context(), "rpcSink(\"localhost\", 34568)"); src.open(); snk.open(); snk.append(new EventImpl("foo".getBytes())); src.next(); snk.close(); src.close(); }