DatabusRelayMain extends HttpRelay { @Override public void pause() { for (Entry<PhysicalPartition, EventProducer> entry : _producers .entrySet()) { EventProducer producer = entry.getValue(); if (null != producer) { if (producer.isRunning()) { producer.pause(); LOG.info("EventProducer :" + producer.getName() + " pause sent"); } else if (producer.isPaused()) { LOG.info("EventProducer :" + producer.getName() + " already paused"); } } } } DatabusRelayMain();  DatabusRelayMain(HttpRelay.Config config,
			PhysicalSourceStaticConfig[] pConfigs);  DatabusRelayMain(HttpRelay.StaticConfig config,
			PhysicalSourceStaticConfig[] pConfigs); void setDbPullerStart(boolean s); boolean getDbPullerStart(); @Override void removeOneProducer(PhysicalSourceStaticConfig pConfig); @Override void addOneProducer(PhysicalSourceStaticConfig pConfig); void initProducers(); MaxSCNReaderWriter getMaxSCNReaderWriter(PhysicalSourceStaticConfig pConfig); static void main(String[] args); @Override void pause(); @Override void resume(); EventProducer[] getProducers(); MonitoringEventProducer[] getMonitoringProducers(); @Override void awaitShutdown(); static final Logger LOG; static final String DB_RELAY_CONFIG_FILE_OPT_NAME; }
@Test public void testRelayChainingSlowConsumer() { DatabusRelayTestUtil.RelayRunner r1=null,r2=null; ClientRunner cr = null; try { String[][] srcNames = { { "com.linkedin.events.example.Account", "com.linkedin.events.example.Settings" }, }; 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(1012, relayPort, 10 * 1024 * 1024, srcConfigs,SCHEMA_REGISTRY_DIR); Assert.assertTrue(null != relay1); r1 = new DatabusRelayTestUtil.RelayRunner(relay1); 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(1013, chainedRelayPort, 1024 * 1024, chainedSrcConfigs,SCHEMA_REGISTRY_DIR); Assert.assertTrue(null != relay2); resetSCN(relay2); r2 = new DatabusRelayTestUtil.RelayRunner(relay2); String srcSubscriptionString = TestUtil.join(srcNames[0], ","); String serverName = "localhost:" + chainedRelayPort; CountingConsumer countingConsumer = new CountingConsumer(500); DatabusSourcesConnection clientConn = RelayEventProducer .createDatabusSourcesConnection("testProducer", serverName, srcSubscriptionString, countingConsumer, 1 * 1024 * 1024, 50000, 30 * 1000, 1000, 15 * 1000, 1, true); cr = new ClientRunner(clientConn); r1.start(); r2.start(); cr.start(); Thread.sleep(20 * 1000); r1.pause(); long maxTimeOutMs = 100 * 1000; long startTime = System.currentTimeMillis(); DbusEventsTotalStats stats = relay1 .getInboundEventStatisticsCollector().getTotalStats(); Assert.assertTrue(stats.getNumSysEvents() > 0); while (countingConsumer.getNumWindows() < stats.getNumSysEvents()) { Thread.sleep(500); 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.getMinScn() < statsChained.getMinScn()); Assert.assertTrue(stats.getNumDataEvents() == statsChained .getNumDataEvents()); Assert.assertTrue(stats.getNumSysEvents() == statsChained .getNumSysEvents()); Assert.assertTrue(countingConsumer.getNumErrors() > 0); 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); } }
@Test public void testRelayChainingPauseResume() { DatabusRelayTestUtil.RelayRunner r1=null,r2=null; ClientRunner cr = null; try { String[][] srcNames = { { "com.linkedin.events.example.Account", "com.linkedin.events.example.Settings" }, }; 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(1014, relayPort, 10 * 1024 * 1024, srcConfigs,SCHEMA_REGISTRY_DIR); Assert.assertTrue(null != relay1); r1 = new DatabusRelayTestUtil.RelayRunner(relay1); 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(1015, chainedRelayPort, 1 * 1024 * 1024, chainedSrcConfigs,SCHEMA_REGISTRY_DIR); Assert.assertTrue(null != relay2); resetSCN(relay2); r2 = new DatabusRelayTestUtil.RelayRunner(relay2); 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); r1.start(); r2.start(); cr.start(); Thread.sleep(2 * 1000); r2.pause(); Thread.sleep(2 * 1000); r1.pause(); r2.unpause(); long maxTimeOutMs = 5 * 1000; long startTime = System.currentTimeMillis(); DbusEventsTotalStats stats = relay1 .getInboundEventStatisticsCollector().getTotalStats(); while (countingConsumer.getNumWindows() < stats.getNumSysEvents()) { Thread.sleep(500); if ((System.currentTimeMillis() - startTime) > maxTimeOutMs) { break; } } 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() == 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); } }
@Test 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" }, }; 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); 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); 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); r1.start(); r2.start(); Thread.sleep(500); cr.start(); r1.pause(); Thread.sleep(10 * 1000); 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()); 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"); } }
@Test 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.info("start"); try { String[][] srcNames = { { "com.linkedin.events.example.Account", "com.linkedin.events.example.Settings" }, }; 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(); 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); 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); r1.start(); Thread.sleep(10*1000); r2.start(); cr.start(); 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); 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); 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); Assert.assertEquals(stats2.getMinScn(), firstGenMinScn) ; Assert.assertEquals(stats2.getMaxScn(), stats3.getMaxScn()); Assert.assertEquals(stats2.getNumSysEvents(), (firstGenChainWindows-1) + stats3.getNumSysEvents()); Assert.assertTrue(stats2.getNumDataEvents() > stats3.getNumDataEvents()); 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"); } }
@Test public void testClientNoEventsResetConnection() throws InterruptedException, InvalidConfigException { LOG.setLevel(Level.ALL); DatabusRelayTestUtil.RelayRunner r1=null; ClientRunner cr = null; try { String srcName = "com.linkedin.events.example.Settings"; int eventRatePerSec = 10; PhysicalSourceConfig srcConfig = DatabusRelayTestUtil.createPhysicalConfigBuilder( (short) 1, DatabusRelayTestUtil.getPhysicalSrcName(srcName), "mock", 500, eventRatePerSec, new String[] {srcName}); int relayPort = 11995; DatabusRelayMain relay = DatabusRelayTestUtil.createDatabusRelayWithSchemaReg(1001, relayPort, 10 * 1024 * 1024, new PhysicalSourceConfig[] {srcConfig},SCHEMA_REGISTRY_DIR); Assert.assertNotEquals(relay, null); r1 = new DatabusRelayTestUtil.RelayRunner(relay); r1.start(); DbusEventsTotalStats stats = relay.getInboundEventStatisticsCollector().getTotalStats(); String srcSubscriptionString = srcName; String serverName = "localhost:" + relayPort; ResetsCountingConsumer countingConsumer = new ResetsCountingConsumer(); DatabusSourcesConnection clientConn = RelayEventProducer .createDatabusSourcesConnection("testProducer", serverName, srcSubscriptionString, countingConsumer, 1 * 1024 * 1024, 50000, 30 * 1000, 100, 15 * 1000, 1, true); cr = new ClientRunner(clientConn); cr.start(); TestUtil.sleep(1000); LOG.info("Sending pause to relay!"); r1.pause(); TestUtil.sleep(4000); LOG.info("no events, time less then threshold. Events=" + countingConsumer.getNumDataEvents() + "; resets = " + countingConsumer.getNumResets()); Assert.assertEquals(countingConsumer.getNumResets(), 0); r1.unpause(); Thread.sleep(8000); LOG.info("some events, more time then timeout. Events=" + countingConsumer.getNumDataEvents() + "; resets = " + countingConsumer.getNumResets()); Assert.assertEquals(countingConsumer.getNumResets(), 0); r1.pause(); clientConn.getRelayPullThread().setNoEventsConnectionResetTimeSec(0); Thread.sleep(8000); LOG.info("no events, more time then timeout, but feature disabled. Events=" + countingConsumer.getNumDataEvents() + "; resets = " + countingConsumer.getNumResets()); Assert.assertEquals(countingConsumer.getNumResets(), 0); clientConn.getRelayPullThread().setNoEventsConnectionResetTimeSec(5); LOG.info("pause the producer. sleep for 6 sec, should reset"); TestUtil.sleep(6000); LOG.info("Client stats=" + countingConsumer); LOG.info("Num resets=" + countingConsumer.getNumResets()); LOG.info("Event windows generated=" + stats.getNumSysEvents()); Assert.assertEquals(countingConsumer.getNumResets(), 0, "0 resets"); Assert.assertEquals(countingConsumer.getNumDataEvents(), stats.getNumDataEvents()); boolean stopped = r1.shutdown(2000); Assert.assertTrue(stopped); LOG.info("Relay r1 stopped"); cr.shutdown(); LOG.info("Client cr stopped"); Assert.assertEquals(countingConsumer.getNumDataEvents(), stats .getNumDataEvents()); } finally { cleanup ( new DatabusRelayTestUtil.RelayRunner[] {r1} , cr); } }
@Test 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" }, }; 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(); 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(); 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(); 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); } }
@Test 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" }, }; 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(); 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(); 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); } }
@Test public void testRelayChainingConnectFailure() { DatabusRelayTestUtil.RelayRunner r1=null,r2=null; ClientRunner cr = null; try { String[][] srcNames = { { "com.linkedin.events.example.fake.FakeSchema", "com.linkedin.events.example.person.Person" }, }; 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(1004, relayPort, 10 * 1024 * 1024, srcConfigs,SCHEMA_REGISTRY_DIR); Assert.assertTrue(null != relay1); r1 = new DatabusRelayTestUtil.RelayRunner(relay1); 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(1005, chainedRelayPort, 10 * 1024 * 1024, chainedSrcConfigs,SCHEMA_REGISTRY_DIR); Assert.assertTrue(null != relay2); r2 = new DatabusRelayTestUtil.RelayRunner(relay2); resetSCN(relay2); 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); r2.start(); cr.start(); Thread.sleep(5 * 1000); r1.start(); Thread.sleep(10 * 1000); r1.pause(); long maxTimeOutMs = 5 * 1000; long startTime = System.currentTimeMillis(); DbusEventsTotalStats stats = relay1 .getInboundEventStatisticsCollector().getTotalStats(); while (countingConsumer.getNumWindows() < stats.getNumSysEvents()) { Thread.sleep(500); if ((System.currentTimeMillis() - startTime) > maxTimeOutMs) { break; } } 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()); } catch (Exception e) { LOG.error("Exception: " + e); Assert.assertTrue(false); } finally { cleanup ( new DatabusRelayTestUtil.RelayRunner[] {r1,r2} , cr); } }
@Test public void testRelalyChainingPartialSubscribe() { DatabusRelayTestUtil.RelayRunner r1=null,r2=null; ClientRunner cr = null; try { String[][] srcNames = { { "com.linkedin.events.example.Account", "com.linkedin.events.example.Settings" }, }; 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(1006, relayPort, 10 * 1024 * 1024, srcConfigs,SCHEMA_REGISTRY_DIR); Assert.assertTrue(null != relay1); r1 = new DatabusRelayTestUtil.RelayRunner(relay1); 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(1007, chainedRelayPort, 10 * 1024 * 1024, chainedSrcConfigs,SCHEMA_REGISTRY_DIR); Assert.assertTrue(null != relay2); resetSCN(relay2); r2 = new DatabusRelayTestUtil.RelayRunner(relay2); String srcSubscriptionString = "com.linkedin.events.example.Settings"; String serverName = "localhost:" + chainedRelayPort; CountingConsumer countingConsumer = new CountingConsumer(); DatabusSourcesConnection clientConn = RelayEventProducer .createDatabusSourcesConnection("testProducer", serverName, srcSubscriptionString, countingConsumer, 1 * 1024 * 1024, -1, 30 * 1000, 100, 15 * 1000, 1, true); cr = new ClientRunner(clientConn); r1.start(); r2.start(); cr.start(); Thread.sleep(10 * 1000); r1.pause(); long maxTimeOutMs = 5 * 1000; long startTime = System.currentTimeMillis(); DbusEventsTotalStats stats = relay1 .getInboundEventStatisticsCollector().getTotalStats(); while (countingConsumer.getNumWindows() < stats.getNumSysEvents()) { Thread.sleep(500); if ((System.currentTimeMillis() - startTime) > maxTimeOutMs) { break; } } 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(countingConsumer.getNumSources() == 1); Assert.assertTrue(stats.getNumSysEvents() == countingConsumer .getNumWindows()); Assert.assertTrue(stats.getNumDataEvents() == 2 * countingConsumer .getNumDataEvents()); } catch (Exception e) { LOG.error("Exception: " + e); Assert.assertTrue(false); } finally { cleanup ( new DatabusRelayTestUtil.RelayRunner[] {r1,r2} , cr); } }
@Test public void testRelayChainingPartialSubscribeRelay() { DatabusRelayTestUtil.RelayRunner r1=null,r2=null; ClientRunner cr = null; try { String[][] srcNames = { { "com.linkedin.events.example.Account", "com.linkedin.events.example.Settings" }, }; 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(1008, relayPort, 10 * 1024 * 1024, srcConfigs,SCHEMA_REGISTRY_DIR); Assert.assertTrue(null != relay1); r1 = new DatabusRelayTestUtil.RelayRunner(relay1); PhysicalSourceConfig[] chainedSrcConfigs = new PhysicalSourceConfig[srcNames.length]; int j = 0; for (String[] srcs : srcNames) { String partialSrcs[] = new String[1]; partialSrcs[0] = srcs[srcs.length - 1]; PhysicalSourceConfig src1 = DatabusRelayTestUtil.createPhysicalConfigBuilder( (short) (srcs.length),DatabusRelayTestUtil.getPhysicalSrcName(srcs[0]), "localhost:" + relayPort, eventRatePerSec, 50, partialSrcs); chainedSrcConfigs[j++] = src1; } int chainedRelayPort = relayPort + 1; DatabusRelayMain relay2 = DatabusRelayTestUtil.createDatabusRelayWithSchemaReg(1009, chainedRelayPort, 10 * 1024 * 1024, chainedSrcConfigs,SCHEMA_REGISTRY_DIR); Assert.assertTrue(null != relay2); resetSCN(relay2); r2 = new DatabusRelayTestUtil.RelayRunner(relay2); String srcSubscriptionString = "com.linkedin.events.example.Settings"; 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); r1.start(); r2.start(); cr.start(); Thread.sleep(10 * 1000); r1.pause(); long maxTimeOutMs = 5 * 1000; long startTime = System.currentTimeMillis(); DbusEventsTotalStats stats = relay1 .getInboundEventStatisticsCollector().getTotalStats(); DbusEventsTotalStats statsChained = relay2 .getInboundEventStatisticsCollector().getTotalStats(); while (countingConsumer.getNumWindows() < stats.getNumSysEvents()) { Thread.sleep(500); if ((System.currentTimeMillis() - startTime) > maxTimeOutMs) { break; } } LOG.info("Client stats=" + countingConsumer); LOG.info("Chained stats=" + statsChained.getNumSysEvents()); LOG.info("Event windows generated=" + stats.getNumSysEvents()); LOG.info("numDataEvents=" + stats.getNumDataEvents() + " numWindows=" + stats.getNumSysEvents() + " size=" + stats.getSizeDataEvents()); Assert.assertTrue(stats.getNumSysEvents() == statsChained .getNumSysEvents()); Assert.assertTrue(statsChained.getNumDataEvents() == countingConsumer .getNumDataEvents()); Assert.assertTrue(stats.getNumDataEvents() == 2 * countingConsumer .getNumDataEvents()); Assert.assertTrue(countingConsumer.getNumSources() == 1); Assert.assertTrue(stats.getNumSysEvents() == countingConsumer .getNumWindows()); } catch (Exception e) { LOG.error("Exception: " + e); Assert.assertTrue(false); } finally { cleanup ( new DatabusRelayTestUtil.RelayRunner[] {r1,r2} , cr); } }
@Test 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" }, }; 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); 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); 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); r1.start(); r2.start(); Thread.sleep(10 * 1000); r1.pause(); cr.start(); long maxTimeOutMs = 15 * 1000; long startTime = System.currentTimeMillis(); DbusEventsTotalStats stats = relay1 .getInboundEventStatisticsCollector().getTotalStats(); while (countingConsumer.getNumWindows() < stats.getNumSysEvents()) { Thread.sleep(500); 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); } }