@Test public void testHeaders() throws InterruptedException, IOException { bind(); Map<String, ByteBuffer> flumeHeaders = new HashMap<>(); flumeHeaders.put("hello", ByteBuffer.wrap("world".getBytes("UTF-8"))); ThriftFlumeEvent thriftEvent = new ThriftFlumeEvent( 1, Priority.INFO, ByteBuffer.wrap("foo".getBytes()), 0, "fooHost", flumeHeaders); FlumeClient fClient = new FlumeClient("0.0.0.0", selectedPort); fClient.append(thriftEvent); Transaction transaction = channel.getTransaction(); transaction.begin(); Event event = channel.take(); Assert.assertNotNull(event); Assert.assertEquals("Event in channel has our header", "world", event.getHeaders().get("hello")); transaction.commit(); transaction.close(); stop(); }
@Override public void stop() { server.stop(); serverTransport.close(); try { thriftHandlerThread.join(); } catch (InterruptedException eI) { LOG.warn("stop interrupted", eI); return; } super.stop(); }
ThriftLegacySource extends AbstractSource implements
    EventDrivenSource, Configurable { @Override public void stop() { server.stop(); serverTransport.close(); try { thriftHandlerThread.join(); } catch (InterruptedException eI) { LOG.warn("stop interrupted", eI); return; } super.stop(); } }
ThriftLegacySource extends AbstractSource implements
    EventDrivenSource, Configurable { @Override public void stop() { server.stop(); serverTransport.close(); try { thriftHandlerThread.join(); } catch (InterruptedException eI) { LOG.warn("stop interrupted", eI); return; } super.stop(); } ThriftLegacySource(); }
ThriftLegacySource extends AbstractSource implements
    EventDrivenSource, Configurable { @Override public void stop() { server.stop(); serverTransport.close(); try { thriftHandlerThread.join(); } catch (InterruptedException eI) { LOG.warn("stop interrupted", eI); return; } super.stop(); } ThriftLegacySource(); @Override void configure(Context context); @SuppressWarnings("deprecation") @Override void start(); @Override void stop(); }
ThriftLegacySource extends AbstractSource implements
    EventDrivenSource, Configurable { @Override public void stop() { server.stop(); serverTransport.close(); try { thriftHandlerThread.join(); } catch (InterruptedException eI) { LOG.warn("stop interrupted", eI); return; } super.stop(); } ThriftLegacySource(); @Override void configure(Context context); @SuppressWarnings("deprecation") @Override void start(); @Override void stop();  }