@Test
/**
 * Chained relay overwrites buffer; consumer starts late; gets only the last n events that fit.
 * Makes sure chained relay overwrites buffer;just like main buffer
 */
public void testRelayChainingDelayedConsumer()
{

	DatabusRelayTestUtil.RelayRunner r1=null,r2=null;
	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 = 10;
		for (String[] srcs : srcNames)
		{
			PhysicalSourceConfig src1 = DatabusRelayTestUtil.createPhysicalConfigBuilder(
					(short) (i + 1),DatabusRelayTestUtil.getPhysicalSrcName(srcs[0]), "mock",
					500, eventRatePerSec, srcs);
			srcConfigs[i++] = src1;
		}
		int relayPort = 11993;
		DatabusRelayMain relay1 = DatabusRelayTestUtil.createDatabusRelayWithSchemaReg(1010, relayPort,
				10 * 1024 * 1024, srcConfigs,SCHEMA_REGISTRY_DIR);
		Assert.assertTrue(null != relay1);
		r1 = new DatabusRelayTestUtil.RelayRunner(relay1);
		// create chained relay with only 1 MB buffer
		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 + 1;
		DatabusRelayMain relay2 = DatabusRelayTestUtil.createDatabusRelayWithSchemaReg(1011,
				chainedRelayPort, 1 * 1024 * 1024, chainedSrcConfigs,SCHEMA_REGISTRY_DIR);
		Assert.assertTrue(null != relay2);
		resetSCN(relay2);
		r2 = new DatabusRelayTestUtil.RelayRunner(relay2);
		// now create client:
		String srcSubscriptionString = TestUtil.join(srcNames[0], ",");
		String serverName = "localhost:" + chainedRelayPort;
		CountingConsumer countingConsumer = new CountingConsumer();
		DatabusSourcesConnection clientConn = RelayEventProducer
				.createDatabusSourcesConnection("testProducer", serverName,
						srcSubscriptionString, countingConsumer,
						1 * 1024 * 1024, 50000, 30 * 1000, 100, 15 * 1000,
						1, true);
		cr = new ClientRunner(clientConn);
		// async starts for all components;
		r1.start();
		// start chained relay
		r2.start();
		// let it run for 10 seconds
		Thread.sleep(10 * 1000);
		r1.pause();
		// start client after the relays have finished
		cr.start();
		// wait until client got all events or for maxTimeout;
		long maxTimeOutMs = 15 * 1000;
		long startTime = System.currentTimeMillis();
		DbusEventsTotalStats stats = relay1
				.getInboundEventStatisticsCollector().getTotalStats();
		while (countingConsumer.getNumWindows() < stats.getNumSysEvents())
		{
			Thread.sleep(500);
			// LOG.info("Client stats=" + countingConsumer);
			// LOG.error("numDataEvents=" +
			// stats.getNumDataEvents() + " numWindows=" +
			// stats.getNumSysEvents() + " size=" +
			// stats.getSizeDataEvents());
			if ((System.currentTimeMillis() - startTime) > maxTimeOutMs)
			{
				break;
			}
		}
		DbusEventsTotalStats statsChained = relay2
				.getInboundEventStatisticsCollector().getTotalStats();
		LOG.info("Client stats=" + countingConsumer);
		LOG.info("Event windows generated="
				+ stats.getNumSysEvents());
		LOG.info("numDataEvents=" + stats.getNumDataEvents()
				+ " numWindows=" + stats.getNumSysEvents() + " size="
				+ stats.getSizeDataEvents());
		Assert.assertTrue(stats.getNumDataEvents() == statsChained
				.getNumDataEvents());
		Assert.assertTrue(stats.getNumSysEvents() == statsChained
				.getNumSysEvents());
		Assert.assertTrue(stats.getNumDataEvents() > countingConsumer
				.getNumDataEvents());
		Assert.assertTrue(countingConsumer.getNumSources() == 2);
		Assert.assertTrue(stats.getNumSysEvents() > countingConsumer
				.getNumWindows());
	}
	catch (Exception e)
	{
		LOG.error("Exception: " + e);
		Assert.assertTrue(false);
	}
	finally
	{
		cleanup ( new DatabusRelayTestUtil.RelayRunner[] {r1,r2} , cr);
	}
}