@Test
/**
 * Test regression of SCN  of chained relay
 */
public void testRelayChainingSCNRegress() throws InvalidConfigException, InterruptedException
{

     final Logger log = Logger.getLogger("TestDatabusRelayMain.testRelayChainingSCNRegress");
	DatabusRelayTestUtil.RelayRunner r1=null,r2=null,r3 = null;
	ClientRunner cr = null;
	//log.setLevel(Level.DEBUG);
	log.info("start");
	//DbusEventBuffer.LOG.setLevel(Level.DEBUG);
	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;
		int largestEventSize = 512*1024;
		long largestWindowSize = 1*1024*1024;
		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;
		final DatabusRelayMain relay1 = DatabusRelayTestUtil.createDatabusRelayWithSchemaReg(1019,
				relayPort, 10 * 1024 * 1024, srcConfigs, SCHEMA_REGISTRY_DIR);
		final DatabusRelayMain relay3 = DatabusRelayTestUtil.createDatabusRelayWithSchemaReg(1020,
				relayPort, 10 * 1024 * 1024, srcConfigs, SCHEMA_REGISTRY_DIR);
		Assert.assertNotNull(relay1);
		Assert.assertNotNull(relay3);
		r1 = new DatabusRelayTestUtil.RelayRunner(relay1);
		final DbusEventsTotalStats stats = relay1.getInboundEventStatisticsCollector().getTotalStats();
		final DbusEventsTotalStats stats3 = relay3.getInboundEventStatisticsCollector().getTotalStats();
		// 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, 500, eventRatePerSec,0,largestEventSize,largestWindowSize,srcs);
			chainedSrcConfigs[j++] = src1;
		}
		int chainedRelayPort = relayPort + 1;
		final DatabusRelayMain relay2 = DatabusRelayTestUtil.createDatabusRelayWithSchemaReg(1021,
				chainedRelayPort, 10 * 1024 * 1024, chainedSrcConfigs,SCHEMA_REGISTRY_DIR);
		Assert.assertNotNull(relay2);
		r2 = new DatabusRelayTestUtil.RelayRunner(relay2);
		resetSCN(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, largestEventSize, 30 * 1000, 100, 15 * 1000,
						1, true,largestEventSize/10);
		cr = new ClientRunner(clientConn);
		// async starts for all components;
		r1.start();
		Thread.sleep(10*1000);
		// start chained relay
		r2.start();
		//start client
		cr.start();
		//Pause r1;
		r1.pause();
		Thread.sleep(1000);
		long firstGenDataEvents = stats.getNumDataEvents();
		long firstGenMinScn  = stats.getMinScn();
		long firstGenWindows = stats.getNumSysEvents();
		Assert.assertTrue(stats.getNumSysEvents() > 0);
		log.warn("relay1:  numDataEvents=" + firstGenDataEvents
				+ " numWindows=" + firstGenWindows + " minScn="
				+ firstGenMinScn + " maxScn=" + stats.getMaxScn());
		Thread.sleep(4*1000);
		//clear the relay
		boolean s = r1.shutdown(2000);
		Assert.assertTrue(s);
		DbusEventsTotalStats stats2 = relay2.getInboundEventStatisticsCollector().getTotalStats();
		long firstGenChainWindows = stats2.getNumSysEvents();
		log.warn("relay2:  numDataEvents=" + stats2.getNumDataEvents()
				+ " numWindows=" + firstGenChainWindows + " minScn="
				+ stats2.getMinScn() + " maxScn=" + stats2.getMaxScn());
		Thread.sleep(2*1000);
		//restart relay
		r3 = new DatabusRelayTestUtil.RelayRunner(relay3);
		r3.start();
		Thread.sleep(15*1000);
		r3.pause();
		Thread.sleep(35*1000);
		log.warn("relay3:  numDataEvents=" + stats3.getNumDataEvents()
				+ " numWindows=" + stats3.getNumSysEvents() + " minScn="
				+ stats3.getMinScn() + " maxScn=" + stats3.getMaxScn());
		stats2 = relay2.getInboundEventStatisticsCollector().getTotalStats();
		log.warn("relay2b: numDataEvents=" + stats2.getNumDataEvents()
				+ " numWindows=" + stats2.getNumSysEvents() + " minScn="
				+ stats2.getMinScn() + " maxScn=" + stats2.getMaxScn());
		log.warn("consumer: " + countingConsumer);
		//compare chained relays with 2 gens of tier 0 relays
		Assert.assertEquals(stats2.getMinScn(), firstGenMinScn) ;
		Assert.assertEquals(stats2.getMaxScn(), stats3.getMaxScn());
		//the total event windows seen by the chained relay will be state of consumption at first failure of relay1 minus 1 overlap window
		Assert.assertEquals(stats2.getNumSysEvents(), (firstGenChainWindows-1) + stats3.getNumSysEvents());
		Assert.assertTrue(stats2.getNumDataEvents() > stats3.getNumDataEvents());
		//compare source to final client
		Assert.assertEquals(countingConsumer.getNumSources(), 2);
		Assert.assertEquals(stats2.getNumSysEvents(), countingConsumer
				.getNumWindows());
		boolean sorted= true;
		long prev = -1;
		log.info(" scn seq on consumer=");
		for (Long l: countingConsumer.getEndScns())
		{
			sorted = sorted && (l >= prev); prev=l;
			log.info(l+ " ");
			if (!sorted) break;
		}
		Assert.assertTrue(sorted);
	}
	finally
	{
		cleanup ( new DatabusRelayTestUtil.RelayRunner[] {r1,r2,r3} , cr);
        log.info("end");
	}
}