@Test
**
* test resetting connection when there is no events for some period of time
* @throws InterruptedException
* @throws InvalidConfigException
*/
public void testClientNoEventsResetConnection() throws InterruptedException, InvalidConfigException
{

 LOG.setLevel(Level.ALL);
  DatabusRelayTestUtil.RelayRunner r1=null;
  ClientRunner cr = null;
  try
  {
    String srcName = "com.linkedin.events.example.Settings";
    // create main relay with random generator
    int eventRatePerSec = 10;
    PhysicalSourceConfig srcConfig = DatabusRelayTestUtil.createPhysicalConfigBuilder(
          (short) 1, DatabusRelayTestUtil.getPhysicalSrcName(srcName), "mock",
          500, eventRatePerSec, new String[] {srcName});
    int relayPort = 11995;
    DatabusRelayMain relay = DatabusRelayTestUtil.createDatabusRelayWithSchemaReg(1001, relayPort,
        10 * 1024 * 1024, new PhysicalSourceConfig[] {srcConfig},SCHEMA_REGISTRY_DIR);
    Assert.assertNotEquals(relay, null);
    r1 = new DatabusRelayTestUtil.RelayRunner(relay);
    // async starts
    r1.start();
    DbusEventsTotalStats stats = relay.getInboundEventStatisticsCollector().getTotalStats();
    // start client in parallel
    String srcSubscriptionString = srcName;
    String serverName = "localhost:" + relayPort;
    ResetsCountingConsumer countingConsumer = new ResetsCountingConsumer();
    DatabusSourcesConnection clientConn = RelayEventProducer
    .createDatabusSourcesConnection("testProducer", serverName,
  		  srcSubscriptionString, countingConsumer,
  		  1 * 1024 * 1024, 50000, 30 * 1000, 100, 15 * 1000,
  		  1, true);
    cr = new ClientRunner(clientConn);
    cr.start();
    TestUtil.sleep(1000); // generate some events
    // pause event generator
    // and wait untill all the events are consumed
    // but since it is less then timeout the connection should NOT reset
    LOG.info("Sending pause to relay!");
    r1.pause();
    TestUtil.sleep(4000);
    LOG.info("no events, time less then threshold. Events=" + countingConsumer.getNumDataEvents() +
  		  "; resets = " + countingConsumer.getNumResets());
    Assert.assertEquals(countingConsumer.getNumResets(), 0);
    // generate more events, more time elapsed then the threshold, but since there are
    // events - NO reset
    r1.unpause();
    Thread.sleep(8000);
    LOG.info("some events, more time then timeout. Events=" + countingConsumer.getNumDataEvents() +
  		  "; resets = " + countingConsumer.getNumResets());
    Assert.assertEquals(countingConsumer.getNumResets(), 0);
    r1.pause(); // stop events
    //set threshold to 0 completely disabling the feature
    clientConn.getRelayPullThread().setNoEventsConnectionResetTimeSec(0);
    Thread.sleep(8000);
    LOG.info("no events, more time then timeout, but feature disabled. Events=" +
  		  countingConsumer.getNumDataEvents() + "; resets = " + countingConsumer.getNumResets());
    Assert.assertEquals(countingConsumer.getNumResets(), 0);
    // enable the feature, and sleep for timeout
    clientConn.getRelayPullThread().setNoEventsConnectionResetTimeSec(5);
    // now wait with no events
    LOG.info("pause the producer. sleep for 6 sec, should reset");
    TestUtil.sleep(6000);
    LOG.info("Client stats=" + countingConsumer);
    LOG.info("Num resets=" + countingConsumer.getNumResets());
    LOG.info("Event windows generated=" + stats.getNumSysEvents());
    Assert.assertEquals(countingConsumer.getNumResets(), 0, "0 resets");
    Assert.assertEquals(countingConsumer.getNumDataEvents(), stats.getNumDataEvents());
    boolean stopped = r1.shutdown(2000);
    Assert.assertTrue(stopped);
    LOG.info("Relay r1 stopped");
    cr.shutdown();
    LOG.info("Client cr stopped");
    Assert.assertEquals(countingConsumer.getNumDataEvents(), stats
  		  .getNumDataEvents());
  }
  finally
  {
    cleanup ( new DatabusRelayTestUtil.RelayRunner[] {r1} , cr);
  }
}