FlumeLog4jAvroAppender extends AppenderSkeleton { @Override public void close() { client = null; } FlumeLog4jAvroAppender(); @Override void close(); @Override boolean requiresLayout(); FlumeEventAvroServer getClient(); void setClient(FlumeEventAvroServer client); String getHostname(); void setHostname(String hostname); int getPort(); void setPort(int port); int getReconnectAttempts(); void setReconnectAttempts(int reconnectAttempts);  }
@Test public void testReconnect() throws IOException { avroLogger.info("message 1"); Event event = eventSource.next(); Assert.assertNotNull(event); Assert.assertEquals("message 1", new String(event.getBody())); eventSource.close(); Callable<Void> logCallable = new Callable<Void>() { @Override public Void call() throws Exception { avroLogger.info("message 2"); return null; } }; ExecutorService logExecutor = Executors.newSingleThreadExecutor(); boolean caughtException = false; try { logExecutor.submit(logCallable); Thread.sleep(1500); eventSource.open(); logExecutor.shutdown(); if (!logExecutor.awaitTermination(10, TimeUnit.SECONDS)) { throw new IllegalStateException( "Log executor is refusing to shutdown cleanly"); } } catch (Throwable t) { logger.error( "Failed to reestablish a connection and log to an avroSource", t); caughtException = true; } Assert.assertFalse(caughtException); event = eventSource.next(); Assert.assertNotNull(event); Assert.assertEquals("message 2", new String(event.getBody())); caughtException = false; try { avroLogger.info("message 3"); } catch (Throwable t) { logger.debug("Logging to a closed server failed (not expected)", t); caughtException = true; } Assert.assertFalse(caughtException); event = eventSource.next(); Assert.assertNotNull(event); Assert.assertEquals("message 3", new String(event.getBody())); }