@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()));
}