@Test
/**
 * DDSDBUS-1904: test a race condition between shutting down a connection and a reconnect to the
 * relay which may cause a Netty connection to be left open after the RelayPuller shutdown.
 * The idea is to trigger a shutdown exactly while we are trying to connect to a relay. We achieve
 * that by rigging MockRelayConnection to call RelayPuller.shutdown() right before responding
 * to requestSources();
 *  */
public void testShutdownRace() throws Exception
{

  final Logger log = Logger.getLogger("TestRelayPullThread.testShutdownRace");
  log.setLevel(Level.INFO);
  log.info("start");
  //elaborate test setup
  List<String> sources = Arrays.asList("source1");
  Properties clientProps = new Properties();
  clientProps.setProperty("client.container.httpPort", "0");
  clientProps.setProperty("client.runtime.bootstrap.enabled", "false");
  clientProps.setProperty("client.runtime.relay(1).name", "relay1");
  clientProps.setProperty("client.runtime.relay(1).port", "10001");
  clientProps.setProperty("client.runtime.relay(1).sources", "source1");
  clientProps.setProperty("client.connectionDefaults.eventBuffer.maxSize", "100000");
  clientProps.setProperty("client.connectionDefaults.pullerRetries.maxRetryNum", "9");
  clientProps.setProperty("client.connectionDefaults.pullerRetries.sleepIncFactor", "1.0");
  clientProps.setProperty("client.connectionDefaults.pullerRetries.sleepIncDelta", "1");
  clientProps.setProperty("client.connectionDefaults.pullerRetries.initSleep", "1");
  DatabusHttpClientImpl client = new DatabusHttpClientImpl("client.", clientProps);
  Assert.assertNotNull(client, "client instantiation ok");
  final DatabusHttpClientImpl.StaticConfig clientConf = client.getClientStaticConfig();
  final DatabusSourcesConnection.StaticConfig srcConnConf = clientConf.getConnectionDefaults();
  DatabusHttpClientImpl.RuntimeConfig clientRtConf = clientConf.getRuntime().build();
  DbusEventBuffer.StaticConfig bufferConf = clientConf.getConnectionDefaults().getEventBuffer();
  DbusEventBuffer relayBuffer = new DbusEventBuffer(bufferConf);
  DbusEventBuffer bootstrapBuffer = new DbusEventBuffer(bufferConf);
  //we keep the index of the next server we expect to see
  AtomicInteger serverIdx = new AtomicInteger(-1);
  Set<ServerInfo> relays = clientRtConf.getRelaysSet();
  //generate the order in which we should see the servers
  List<ServerInfo> relayOrder = new ArrayList<ServerInfo>(relays);
  if (LOG.isInfoEnabled())
  {
    StringBuilder sb = new StringBuilder();
    for (ServerInfo serverInfo: relayOrder)
    {
      sb.append(serverInfo.getName());
      sb.append(" ");
    }
    LOG.info("Relay order:" + sb.toString());
  }
  List<IdNamePair> sourcesResponse = new ArrayList<IdNamePair>();
  sourcesResponse.add(new IdNamePair(1L, "source1"));
  RegisterResponseEntry rre1 = new RegisterResponseEntry(1L, (short)1, SCHEMA$.toString());
  final HashMap<Long, List<RegisterResponseEntry>> registerResponse =
      new HashMap<Long, List<RegisterResponseEntry>>();
  registerResponse.put(1L, Arrays.asList(rre1));
  //This guy succeeds on both /sources and /register
  final MockRelayConnection relayConn1 =
      new MockRelayConnection(sourcesResponse,
                              registerResponse,
                              null,
                              serverIdx);
  //This guy will succeed on /sources but will force a shutdown so that
  //the success message is ignored
  final MockRelayConnection relayConn2 =
      new MockRelayConnectionForTestShutdownRace(sourcesResponse,
                                                 null,
                                                 null,
                                                 serverIdx,
                                                 log);
  DatabusRelayConnectionFactory mockConnFactory =
      EasyMock.createMock("mockRelayFactory", DatabusRelayConnectionFactory.class);
  // expected scenario:  create relayConn1 -> /sources success -> /register success ->
  // /stream error -> create relayConn2 -> call /sources -> shut down puller -> return
  // success for /sources
  EasyMock.expect(mockConnFactory.createRelayConnection(
      serverNameMatcher(serverIdx, relayOrder),
      EasyMock.<ActorMessageQueue>notNull(),
      EasyMock.<RemoteExceptionHandler>notNull())).andReturn(relayConn1);
  EasyMock.expect(mockConnFactory.createRelayConnection(
      serverNameMatcher(serverIdx, relayOrder),
      EasyMock.<ActorMessageQueue>notNull(),
      EasyMock.<RemoteExceptionHandler>notNull())).andReturn(relayConn2);
  EasyMock.replay(mockConnFactory);
  List<DatabusSubscription> sourcesSubList = DatabusSubscription.createSubscriptionList(sources);
  //Dummy connection object as expected by the puller thread
  // Note that in this case, it is ok to pass Set<relays> as all the relays serve the same source "source1"
  ConnectionStateFactory connStateFactory = new ConnectionStateFactory(sources);
  DatabusSourcesConnection sourcesConn = new DatabusSourcesConnection(
      srcConnConf, sourcesSubList, relays, null, null, null, relayBuffer, bootstrapBuffer,
      Executors.newCachedThreadPool(), null, null, null, null, null, null, null, mockConnFactory, null,
      null, null, null, new DbusEventV1Factory(), connStateFactory);
  final RelayPullThread relayPuller =
      new RelayPullThread("RelayPuller", sourcesConn, relayBuffer, connStateFactory, relays,
                          new ArrayList<DbusKeyCompositeFilterConfig>(),
                          !clientConf.getRuntime().getBootstrap().isEnabled(),
                          clientConf.isReadLatestScnOnErrorEnabled(),
                          clientConf.getPullerBufferUtilizationPct(),
                          Integer.MAX_VALUE,
                          ManagementFactory.getPlatformMBeanServer(),
                          new DbusEventV1Factory(),
                          null);
  //relayPuller.getLog().setLevel(Level.INFO);
  RemoteExceptionHandler mockRemoteExceptionHandler =
      new MockRemoteExceptionHandler(sourcesConn, relayBuffer, relayPuller);
  Field field = relayPuller.getClass().getDeclaredField("_remoteExceptionHandler");
  field.setAccessible(true);
  field.set(relayPuller, mockRemoteExceptionHandler);
  relayConn1.setCallback(relayPuller);
  relayConn2.setCallback(relayPuller);
  //Let the show begin
  final Thread relayPullerThread = new Thread(relayPuller, "testShutdownRace.RelayPuller");
  relayPullerThread.setDaemon(true);
  relayPullerThread.start();
  relayPuller.enqueueMessage(LifecycleMessage.createStartMessage());
  //wait for the puller to go the STREAM_REQUEST_SUCCESS state
  TestUtil.assertWithBackoff(new ConditionCheck()
  {
    @Override
    public boolean check()
    {
      return null != relayConn1.getLastStateMsg();
    }
  }, "wait for call from the puller to the relay connection", 500, log);
  //wait for puller thread to shutdown
  TestUtil.assertWithBackoff(new ConditionCheck()
  {
    @Override
    public boolean check()
    {
      log.debug(relayPuller.getMessageHistoryLog());
      return !relayPullerThread.isAlive();
    }
  }, "wait for puller to shutdown", 1000, log);
  EasyMock.verify(mockConnFactory);
  Assert.assertEquals(relayPuller.getLastOpenConnection(), null);
  log.info("done");
}