@Test
/**
 * Test relay chaining that utilizes restart SCN offset
 */
public void testRelayChainingRestartSCNOffset() throws InterruptedException, InvalidConfigException
{

  final Logger log = Logger.getLogger("TestDatabusRelayMain.testRelayChainingRestartSCNOffset");
  log.info("start");
	DatabusRelayTestUtil.RelayRunner r1=null,r2=null,r3=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 = 2;
		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(1016, relayPort,
				10 * 1024 * 1024, srcConfigs,SCHEMA_REGISTRY_DIR);
		Assert.assertTrue(null != relay1);
		r1 = new DatabusRelayTestUtil.RelayRunner(relay1);
		// 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 + 1;
		DatabusRelayMain relay2 = DatabusRelayTestUtil.createDatabusRelayWithSchemaReg(1017,
				chainedRelayPort, 1 * 1024 * 1024, chainedSrcConfigs,SCHEMA_REGISTRY_DIR);
		Assert.assertTrue(null != relay2);
		r2 = new DatabusRelayTestUtil.RelayRunner(relay2);
		resetSCN(relay2);
		// now create client:
		String srcSubscriptionString = TestUtil.join(srcNames[0], ",");
		String serverName = "localhost:" + chainedRelayPort;
		final 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();
		// start client
		Thread.sleep(500);
		cr.start();
		r1.pause();
		// let it run for 10 seconds
		Thread.sleep(10 * 1000);
		// wait until client got all events or for maxTimeout;
		final long maxTimeOutMs = 5 * 1000;
		final DbusEventsTotalStats dbRelayStats = relay1
				.getInboundEventStatisticsCollector().getTotalStats();
		TestUtil.assertWithBackoff(new ConditionCheck() {
			@Override
			public boolean check() {
				return countingConsumer.getNumWindows() == dbRelayStats.getNumSysEvents();
			}
		}, "consumer caught up", maxTimeOutMs, log);
		log.info("Client stats=" + countingConsumer);
		log.info("Event windows generated="
				+ dbRelayStats.getNumSysEvents());
		log.info("numDataEvents=" + dbRelayStats.getNumDataEvents()
				+ " numWindows=" + dbRelayStats.getNumSysEvents() + " size="
				+ dbRelayStats.getSizeDataEvents());
		Assert.assertEquals(dbRelayStats.getNumDataEvents(), countingConsumer
				.getNumDataEvents());
		Assert.assertEquals(countingConsumer.getNumSources(), 2);
		Assert.assertEquals(dbRelayStats.getNumSysEvents(), countingConsumer
				.getNumWindows());
		cr.shutdown();
		boolean s2= r2.shutdown(2000);
		Assert.assertTrue(s2);
		Assert.assertTrue(!r2.isAlive());
		//start r3; new chained relay with restart SCN offset; we get some data;
		chainedSrcConfigs[0].setRestartScnOffset(dbRelayStats.getMaxScn() - dbRelayStats.getPrevScn());
		DatabusRelayMain relay3 = DatabusRelayTestUtil.createDatabusRelayWithSchemaReg(1018,
				chainedRelayPort, 1 * 1024 * 1024, chainedSrcConfigs,SCHEMA_REGISTRY_DIR);
		r3 = new DatabusRelayTestUtil.RelayRunner(relay3);
		r3.start();
		final DbusEventsTotalStats newChainedRlyStats = relay3
				.getInboundEventStatisticsCollector().getTotalStats();
		TestUtil.assertWithBackoff(new ConditionCheck() {
			@Override
			public boolean check() {
				return newChainedRlyStats.getNumDataEvents() > 0;
			}
		}, "new chained relay running", 5000, log);
		log.info("Stats3= numDataEvents=" + newChainedRlyStats.getNumDataEvents()
				+ " numWindows=" + newChainedRlyStats.getNumSysEvents() + " size="
				+ newChainedRlyStats.getSizeDataEvents());
	}
	finally
	{
		cleanup ( new DatabusRelayTestUtil.RelayRunner[] {r1,r2,r3} , cr);
           log.info("end");
	}
}