@Test
/**
 * Concurrent consumption of chained relay by client and by chained relay from regular relay
 */
public void testRelayChainingBasic() throws InterruptedException, InvalidConfigException
{

	DatabusRelayTestUtil.RelayRunner r1=null,r2=null;
       final Logger log = Logger.getLogger("TestDatabusRelayMain.testRelayChainingBasic");
       Logger.getRootLogger().setLevel(Level.INFO);
       log.setLevel(Level.DEBUG);
       log.debug("available processors:" + Runtime.getRuntime().availableProcessors());
       log.debug("available memory:" + Runtime.getRuntime().freeMemory());
       log.debug("total memory:" + Runtime.getRuntime().totalMemory());
	ClientRunner cr = null;
	try
	{
		String[][] srcNames =
		{
            { "com.linkedin.events.example.fake.FakeSchema",
              "com.linkedin.events.example.person.Person" }, };
		// 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("create chained relay");
		PhysicalSourceConfig[] chainedSrcConfigs = new PhysicalSourceConfig[srcNames.length];
		int j = 0;
		for (String[] srcs : srcNames)
		{
			PhysicalSourceConfig src1 = DatabusRelayTestUtil.createPhysicalConfigBuilder(
					(short) (j + 1), DatabusRelayTestUtil.getPhysicalSrcName(srcs[0]),
					"localhost:" + relayPort, eventRatePerSec, 50, srcs);
			chainedSrcConfigs[j++] = src1;
		}
		int chainedRelayPort = relayPort + 1000;
		DatabusRelayMain relay2 = DatabusRelayTestUtil.createDatabusRelayWithSchemaReg(1003,
				chainedRelayPort, 1 * 1024 * 1024, chainedSrcConfigs, SCHEMA_REGISTRY_DIR);
		Assert.assertNotNull(relay2);
		r2 = new DatabusRelayTestUtil.RelayRunner(relay2);
		resetSCN(relay2);
		log.info("now create client");
		String srcSubscriptionString = TestUtil.join(srcNames[0], ",");
		String serverName = "localhost:" + chainedRelayPort;
		final CountingConsumer countingConsumer = new CountingConsumer();
		//Set maxSize to 100K, maxEventSize to 50K
		DatabusSourcesConnection clientConn = RelayEventProducer
				.createDatabusSourcesConnection("testProducer", serverName,
						srcSubscriptionString, countingConsumer,
						100 * 1024 , 10000, 30 * 1000, 100, 30 * 1000,
						1, true);
		cr = new ClientRunner(clientConn);
		log.info("async starts for all components");
		r1.start();
		assertRelayRunning(r1.getRelay(), 5000, log);
		log.info("start chained relay");
		r2.start();
           assertRelayRunning(r2.getRelay(), 5000, log);
		Thread.sleep(5*1000);
		r1.pause();
     // wait for r2 to catchup with r1
     final DbusEventsTotalStats stats = relay1
         .getInboundEventStatisticsCollector().getTotalStats();
     final DbusEventsTotalStats stats2 = relay2
         .getInboundEventStatisticsCollector().getTotalStats();
     TestUtil.assertWithBackoff(new ConditionCheck()
     {
       @Override
       public boolean check()
       {
         log.debug("stats2.getNumSysEvents()=" + stats2.getNumSysEvents());
         return stats2.getNumSysEvents() == stats.getNumSysEvents();
       }
     }, "wait for chained relay to catchup", 60000, log);
     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());
		LOG.info("numDataEvents2=" + stats2.getNumDataEvents()
				+ " numWindows2=" + stats2.getNumSysEvents() + " size2="
				+ stats2.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);
	}
}