@Test
/**
 * Basic consumption client and by chained relay from regular relay
 * but start with small readBuffer (20K) - Set up large events
 */
public void testDynamicBufferGrowthClient() throws InterruptedException, InvalidConfigException
{

  DatabusRelayTestUtil.RelayRunner r1=null,r2=null;
  final Logger log = Logger.getLogger("TestDatabusRelayMain.testDynamicBufferGrowth");
  Logger.getRootLogger().setLevel(Level.INFO);
  ClientRunner cr = null;
  try
  {
    String[][] srcNames =
    {
            { "com.linkedin.events.example.Account",
              "com.linkedin.events.example.Settings" }, };
    // create main relay with random generator
    PhysicalSourceConfig[] srcConfigs = new PhysicalSourceConfig[srcNames.length];
    int i = 0;
    int eventRatePerSec = 20;
    for (String[] srcs : srcNames)
    {
      PhysicalSourceConfig src1 = DatabusRelayTestUtil.createPhysicalConfigBuilder(
          (short) (i + 1), DatabusRelayTestUtil.getPhysicalSrcName(srcs[0]), "mock",
          100, eventRatePerSec, srcs);
      srcConfigs[i++] = src1;
    }
    int relayPort = Utils.getAvailablePort(11994);
    DatabusRelayMain relay1 = DatabusRelayTestUtil.createDatabusRelayWithSchemaReg(1002, relayPort,
        10 * 1024 * 1024, srcConfigs, SCHEMA_REGISTRY_DIR);
    Assert.assertNotNull(relay1);
    r1 = new DatabusRelayTestUtil.RelayRunner(relay1);
    log.info("now create client");
    String srcSubscriptionString = TestUtil.join(srcNames[0], ",");
    String serverName = "localhost:" + relayPort;
    final CountingConsumer countingConsumer = new CountingConsumer();
    //Set maxSize to 100K, maxEventSize to 15k, set init readBufferSize to maxEventSize/2
    int maxEventSize=15*1024;
    int initReadBufferSize=maxEventSize/2;
    DatabusSourcesConnection clientConn = RelayEventProducer
        .createDatabusSourcesConnection("testProducer", serverName,
            srcSubscriptionString, countingConsumer,
            100 * 1024 , maxEventSize, 30 * 1000, 100, 30 * 1000,
            1, true,initReadBufferSize);
    cr = new ClientRunner(clientConn);
    log.info("async starts for all components");
    r1.start();
    assertRelayRunning(r1.getRelay(), 5000, log);
    Thread.sleep(5*1000);
    r1.pause();
    final DbusEventsTotalStats stats = relay1
        .getInboundEventStatisticsCollector().getTotalStats();
    log.info("start the client");
    cr.start();
    // wait until client got all events or for maxTimeout;
    TestUtil.assertWithBackoff(new ConditionCheck()
    {
      @Override
      public boolean check()
      {
        log.debug("countingConsumer.getNumWindows()="
            + countingConsumer.getNumWindows());
        return countingConsumer.getNumWindows() == stats.getNumSysEvents();
      }
    }, "wait until client got all events", 10000, log);
    LOG.info("Client stats=" + countingConsumer);
    LOG.info("Event windows generated="
        + stats.getNumSysEvents());
    LOG.info("numDataEvents=" + stats.getNumDataEvents()
        + " numWindows=" + stats.getNumSysEvents() + " size="
        + stats.getSizeDataEvents());
    Assert.assertEquals(stats.getNumDataEvents(), countingConsumer
        .getNumDataEvents());
    Assert.assertEquals(countingConsumer.getNumSources(), 2);
    Assert.assertEquals(stats.getNumSysEvents(), countingConsumer
        .getNumWindows());
  }
  finally
  {
    cleanup ( new DatabusRelayTestUtil.RelayRunner[] {r1,r2} , cr);
  }
}