MultiConsumerCallback implements DatabusStreamConsumer { public void setSourceMap(Map<Long, IdNamePair> sourceMap) { _sourceMap = sourceMap; } MultiConsumerCallback(List<DatabusV2ConsumerRegistration> registrations,
                               ExecutorService executorService,
                               long timeBudgetMs,
                               ConsumerCallbackFactory<DatabusCombinedConsumer> callbackFactory);  MultiConsumerCallback(List<DatabusV2ConsumerRegistration> registrations,
                               ExecutorService executorService,
                               long timeBudgetMs,
                               ConsumerCallbackFactory<DatabusCombinedConsumer> callbackFactory,
                               ConsumerCallbackStats consumerStats,    // specific to relay or bootstrap mode, not both
                               UnifiedClientStats unifiedClientStats,  // used in both relay and bootstrap mode
                               LoggingConsumer loggingConsumer,
                               Logger log); ConsumerCallbackResult flushCallQueue(long curTime); @Override ConsumerCallbackResult onCheckpoint(SCN checkpointScn); @Override ConsumerCallbackResult onDataEvent(DbusEvent e, DbusEventDecoder eventDecoder); @Override ConsumerCallbackResult onEndDataEventSequence(SCN endScn); @Override ConsumerCallbackResult onEndSource(String source, Schema sourceSchema); @Override ConsumerCallbackResult onRollback(SCN startScn); @Override ConsumerCallbackResult onStartDataEventSequence(SCN startScn); @Override ConsumerCallbackResult onStartSource(String source, Schema sourceSchema); @Override ConsumerCallbackResult onStartConsumption(); @Override ConsumerCallbackResult onStopConsumption(); void setSourceMap(Map<Long, IdNamePair> sourceMap); Map<Long, IdNamePair> getSourceMap(); @Override ConsumerCallbackResult onError(Throwable err); ConsumerCallbackStats getStats(); void removeRegistration(DatabusV2ConsumerRegistration reg); final String MODULE; public Logger _log; }
@Test(groups = {"small", "functional"}) public void test1StreamConsumerHappyPath() { LOG.info("\n\nstarting test1StreamConsumerHappyPath()"); Hashtable<Long, AtomicInteger> keyCounts = new Hashtable<Long, AtomicInteger>(); DbusEventBuffer eventsBuf = new DbusEventBuffer(_generic100KBufferStaticConfig); eventsBuf.start(0); eventsBuf.startEvents(); initBufferWithEvents(eventsBuf, 1, 1, (short)1, keyCounts); initBufferWithEvents(eventsBuf, 2, 2, (short) 3, keyCounts); eventsBuf.endEvents(100L); DatabusStreamConsumer mockConsumer = EasyMock.createStrictMock(DatabusStreamConsumer.class); SelectingDatabusCombinedConsumer sdccMockConsumer = new SelectingDatabusCombinedConsumer(mockConsumer); List<String> sources = new ArrayList<String>(); Map<Long, IdNamePair> sourcesMap = new HashMap<Long, IdNamePair>(); for (int i = 1; i <= 3; ++i) { IdNamePair sourcePair = new IdNamePair((long)i, "source" + i); sources.add(sourcePair.getName()); sourcesMap.put(sourcePair.getId(), sourcePair); } DatabusV2ConsumerRegistration consumerReg = new DatabusV2ConsumerRegistration(sdccMockConsumer, sources, null); List<DatabusV2ConsumerRegistration> allRegistrations = Arrays.asList(consumerReg); MultiConsumerCallback callback = new MultiConsumerCallback( allRegistrations, Executors.newCachedThreadPool(), 60000, new StreamConsumerCallbackFactory(null, null), null, null, null, null); callback.setSourceMap(sourcesMap); DbusEventBuffer.DbusEventIterator iter = eventsBuf.acquireIterator("myIter1"); assert iter.hasNext() : "unable to read event"; DbusEvent event1 = iter.next(); assert iter.hasNext() : "unable to read event"; DbusEvent event2 = iter.next(); assert iter.hasNext() : "unable to read event"; DbusEvent event3 = iter.next(); initMockStreamConsumer3EventFullLifecycle(mockConsumer, event1, event2, event3, keyCounts); assert3EventFullLifecycle(callback, event1, event2, event3); EasyMock.verify(mockConsumer); assert keyCounts.get(1L).get() == 1 : "invalid number of event(1) calls: " + keyCounts.get(1L).get(); assert keyCounts.get(2L).get() == 1 : "invalid number of event(2) calls:" + keyCounts.get(2L).get(); assert keyCounts.get(3L).get() == 1 : "invalid number of event(3) calls:" + keyCounts.get(3L).get(); }
@Test(groups = {"small", "functional"}) public void test1StreamConsumerCallFailure() { LOG.info("\n\nstarting test1StreamConsumerCallFailure()"); Hashtable<Long, AtomicInteger> keyCounts = new Hashtable<Long, AtomicInteger>(); DbusEventBuffer eventsBuf = new DbusEventBuffer(_generic100KBufferStaticConfig); eventsBuf.start(0); eventsBuf.startEvents(); initBufferWithEvents(eventsBuf, 1, 1, (short)1, keyCounts); initBufferWithEvents(eventsBuf, 2, 2, (short)3, keyCounts); eventsBuf.endEvents(100L); DatabusStreamConsumer mockConsumer = EasyMock.createStrictMock(DatabusStreamConsumer.class); SelectingDatabusCombinedConsumer sdccMockConsumer = new SelectingDatabusCombinedConsumer(mockConsumer); List<String> sources = new ArrayList<String>(); Map<Long, IdNamePair> sourcesMap = new HashMap<Long, IdNamePair>(); for (int i = 1; i <= 3; ++i) { IdNamePair sourcePair = new IdNamePair((long)i, "source" + i); sources.add(sourcePair.getName()); sourcesMap.put(sourcePair.getId(), sourcePair); } DatabusV2ConsumerRegistration consumerReg = new DatabusV2ConsumerRegistration(sdccMockConsumer, sources, null); List<DatabusV2ConsumerRegistration> allRegistrations = Arrays.asList(consumerReg); MultiConsumerCallback callback = new MultiConsumerCallback( allRegistrations, Executors.newCachedThreadPool(), 1000, new StreamConsumerCallbackFactory(null, null), null, null, null, null); callback.setSourceMap(sourcesMap); DbusEventBuffer.DbusEventIterator iter = eventsBuf.acquireIterator("myIter1"); assert iter.hasNext() : "unable to read event"; DbusEvent event1 = iter.next(); assert iter.hasNext() : "unable to read event"; DbusEvent event2 = iter.next(); assert iter.hasNext() : "unable to read event"; DbusEvent event3 = iter.next(); initMockFailingStreamConsumer3EventFullLifecycle(mockConsumer, event1, event2, event3, keyCounts); assert3EventFullLifecycleWithFailure(callback, event1, event2, event3); EasyMock.verify(mockConsumer); assert keyCounts.get(1L).get() == 1 : "invalid number of event(1) calls: " + keyCounts.get(1L).get(); assert keyCounts.get(2L).get() == 1 : "invalid number of event(2) calls:" + keyCounts.get(2L).get(); assert keyCounts.get(3L).get() == 1 : "invalid number of event(3) calls:" + keyCounts.get(3L).get(); }
@Test(groups = {"small", "functional"}) public void test3IndependentStreamConsumersHappyPath() { LOG.info("\n\nstarting test3IndependentStreamConsumersHappyPath()"); Hashtable<Long, AtomicInteger> keyCounts = new Hashtable<Long, AtomicInteger>(); DbusEventBuffer eventsBuf = new DbusEventBuffer(_generic100KBufferStaticConfig); eventsBuf.start(0); eventsBuf.startEvents(); initBufferWithEvents(eventsBuf, 1, 1, (short)1, keyCounts); initBufferWithEvents(eventsBuf, 2, 2, (short)3, keyCounts); eventsBuf.endEvents(100L); DatabusStreamConsumer mockConsumer1 = EasyMock.createStrictMock("consumer1", DatabusStreamConsumer.class); SelectingDatabusCombinedConsumer sdccMockConsumer1 = new SelectingDatabusCombinedConsumer(mockConsumer1); EasyMock.makeThreadSafe(mockConsumer1, true); DatabusStreamConsumer mockConsumer2 = EasyMock.createStrictMock("consumer2", DatabusStreamConsumer.class); SelectingDatabusCombinedConsumer sdccMockConsumer2 = new SelectingDatabusCombinedConsumer(mockConsumer2); EasyMock.makeThreadSafe(mockConsumer2, true); DatabusStreamConsumer mockConsumer3 = EasyMock.createStrictMock("consumer3", DatabusStreamConsumer.class); SelectingDatabusCombinedConsumer sdccMockConsumer3 = new SelectingDatabusCombinedConsumer(mockConsumer3); EasyMock.makeThreadSafe(mockConsumer3, true); List<String> sources = new ArrayList<String>(); Map<Long, IdNamePair> sourcesMap = new HashMap<Long, IdNamePair>(); for (int i = 1; i <= 3; ++i) { IdNamePair sourcePair = new IdNamePair((long)i, "source" + i); sources.add(sourcePair.getName()); sourcesMap.put(sourcePair.getId(), sourcePair); } DatabusV2ConsumerRegistration consumerReg1 = new DatabusV2ConsumerRegistration(sdccMockConsumer1, sources, null); DatabusV2ConsumerRegistration consumerReg2 = new DatabusV2ConsumerRegistration(sdccMockConsumer2, sources, null); DatabusV2ConsumerRegistration consumerReg3 = new DatabusV2ConsumerRegistration(sdccMockConsumer3, sources, null); List<DatabusV2ConsumerRegistration> allRegistrations = Arrays.asList(consumerReg1, consumerReg2, consumerReg3); MultiConsumerCallback callback = new MultiConsumerCallback( allRegistrations, Executors.newCachedThreadPool(), 1000, new StreamConsumerCallbackFactory(null, null), null, null, null, null); callback.setSourceMap(sourcesMap); DbusEventBuffer.DbusEventIterator iter = eventsBuf.acquireIterator("myIter1"); assert iter.hasNext() : "unable to read event"; DbusEvent event1 = iter.next(); assert iter.hasNext() : "unable to read event"; DbusEvent event2 = iter.next(); assert iter.hasNext() : "unable to read event"; DbusEvent event3 = iter.next(); initMockStreamConsumer3EventFullLifecycle(mockConsumer1, event1, event2, event3, keyCounts); initMockStreamConsumer3EventFullLifecycle(mockConsumer2, event1, event2, event3, keyCounts); initMockStreamConsumer3EventFullLifecycle(mockConsumer3, event1, event2, event3, keyCounts); assert3EventFullLifecycle(callback, event1, event2, event3); EasyMock.verify(mockConsumer1); EasyMock.verify(mockConsumer2); EasyMock.verify(mockConsumer3); assert keyCounts.get(1L).get() == 3 : "invalid number of event(1) calls: " + keyCounts.get(1L).get(); assert keyCounts.get(2L).get() == 3 : "invalid number of event(2) calls:" + keyCounts.get(2L).get(); assert keyCounts.get(3L).get() == 3 : "invalid number of event(3) calls:" + keyCounts.get(3L).get(); }
@Test(groups = {"small", "functional"}) public void test3IndependentStreamConsumersWithFailure() { LOG.info("\n\nstarting test3IndependentStreamConsumersWithFailure()"); Hashtable<Long, AtomicInteger> keyCounts = new Hashtable<Long, AtomicInteger>(); DbusEventBuffer eventsBuf = new DbusEventBuffer(_generic100KBufferStaticConfig); eventsBuf.start(0); eventsBuf.startEvents(); initBufferWithEvents(eventsBuf, 1, 1, (short)1, keyCounts); initBufferWithEvents(eventsBuf, 2, 2, (short)3, keyCounts); eventsBuf.endEvents(100L); DatabusStreamConsumer mockConsumer1 = EasyMock.createStrictMock("consumer1", DatabusStreamConsumer.class); EasyMock.makeThreadSafe(mockConsumer1, true); DatabusStreamConsumer mockConsumer2 = EasyMock.createStrictMock("consumer2", DatabusStreamConsumer.class); EasyMock.makeThreadSafe(mockConsumer2, true); DatabusStreamConsumer mockConsumer3 = EasyMock.createStrictMock("consumer3", DatabusStreamConsumer.class); EasyMock.makeThreadSafe(mockConsumer3, true); SelectingDatabusCombinedConsumer sdccMockConsumer1 = new SelectingDatabusCombinedConsumer(mockConsumer1); SelectingDatabusCombinedConsumer sdccMockConsumer2 = new SelectingDatabusCombinedConsumer(mockConsumer2); SelectingDatabusCombinedConsumer sdccMockConsumer3 = new SelectingDatabusCombinedConsumer(mockConsumer3); List<String> sources = new ArrayList<String>(); Map<Long, IdNamePair> sourcesMap = new HashMap<Long, IdNamePair>(); for (int i = 1; i <= 3; ++i) { IdNamePair sourcePair = new IdNamePair((long)i, "source" + i); sources.add(sourcePair.getName()); sourcesMap.put(sourcePair.getId(), sourcePair); } DatabusV2ConsumerRegistration consumerReg1 = new DatabusV2ConsumerRegistration(sdccMockConsumer1, sources, null); DatabusV2ConsumerRegistration consumerReg2 = new DatabusV2ConsumerRegistration(sdccMockConsumer2, sources, null); DatabusV2ConsumerRegistration consumerReg3 = new DatabusV2ConsumerRegistration(sdccMockConsumer3, sources, null); List<DatabusV2ConsumerRegistration> allRegistrations = Arrays.asList(consumerReg1, consumerReg2, consumerReg3); MultiConsumerCallback callback = new MultiConsumerCallback( allRegistrations, Executors.newCachedThreadPool(), 1000, new StreamConsumerCallbackFactory(null, null), null, null, null, null); callback.setSourceMap(sourcesMap); DbusEventBuffer.DbusEventIterator iter = eventsBuf.acquireIterator("myIter1"); assert iter.hasNext() : "unable to read event"; DbusEvent event1 = iter.next(); assert iter.hasNext() : "unable to read event"; DbusEvent event2 = iter.next(); assert iter.hasNext() : "unable to read event"; DbusEvent event3 = iter.next(); initMockStreamConsumer3EventFullLifecycle(mockConsumer1, event1, event2, event3, keyCounts); initMockFailingStreamConsumer3EventFullLifecycle(mockConsumer2, event1, event2, event3, keyCounts); initMockStreamConsumer3EventFullLifecycle(mockConsumer3, event1, event2, event3, keyCounts); assert3EventFullLifecycleWithFailure(callback, event1, event2, event3); EasyMock.verify(mockConsumer1); EasyMock.verify(mockConsumer2); EasyMock.verify(mockConsumer3); assert keyCounts.get(1L).get() == 3 : "invalid number of event(1) calls: " + keyCounts.get(1L).get(); assert keyCounts.get(2L).get() == 3 : "invalid number of event(2) calls:" + keyCounts.get(2L).get(); assert keyCounts.get(3L).get() == 3 : "invalid number of event(3) calls:" + keyCounts.get(3L).get(); }
@Test(groups = {"small", "functional"}) public void test3GroupedStreamConsumersHappyPath() { LOG.info("\n\nstarting test3GroupedStreamConsumersHappyPath()"); Hashtable<Long, AtomicInteger> keyCounts = new Hashtable<Long, AtomicInteger>(); DbusEventBuffer eventsBuf = new DbusEventBuffer(_generic100KBufferStaticConfig); eventsBuf.start(0); eventsBuf.startEvents(); initBufferWithEvents(eventsBuf, 1, 1, (short)1, keyCounts); initBufferWithEvents(eventsBuf, 2, 2, (short)3, keyCounts); eventsBuf.endEvents(100L); DatabusStreamConsumer mockConsumer1 = EasyMock.createStrictMock("consumer1", DatabusStreamConsumer.class); EasyMock.makeThreadSafe(mockConsumer1, true); DatabusStreamConsumer mockConsumer2 = EasyMock.createStrictMock("consumer2", DatabusStreamConsumer.class); EasyMock.makeThreadSafe(mockConsumer2, true); DatabusStreamConsumer mockConsumer3 = EasyMock.createStrictMock("consumer3", DatabusStreamConsumer.class); EasyMock.makeThreadSafe(mockConsumer3, true); DatabusCombinedConsumer sdccMockConsumer1 = new SelectingDatabusCombinedConsumer(mockConsumer1); DatabusCombinedConsumer sdccMockConsumer2 = new SelectingDatabusCombinedConsumer(mockConsumer2); DatabusCombinedConsumer sdccMockConsumer3 = new SelectingDatabusCombinedConsumer(mockConsumer3); List<String> sources = new ArrayList<String>(); Map<Long, IdNamePair> sourcesMap = new HashMap<Long, IdNamePair>(); for (int i = 1; i <= 3; ++i) { IdNamePair sourcePair = new IdNamePair((long)i, "source" + i); sources.add(sourcePair.getName()); sourcesMap.put(sourcePair.getId(), sourcePair); } DatabusV2ConsumerRegistration consumerReg1 = new DatabusV2ConsumerRegistration( Arrays.asList(sdccMockConsumer1, sdccMockConsumer2, sdccMockConsumer3), sources, null); List<DatabusV2ConsumerRegistration> allRegistrations = Arrays.asList(consumerReg1); MultiConsumerCallback callback = new MultiConsumerCallback( allRegistrations, Executors.newCachedThreadPool(), 1000, new StreamConsumerCallbackFactory(null, null), null, null, null, null); callback.setSourceMap(sourcesMap); DbusEventBuffer.DbusEventIterator iter = eventsBuf.acquireIterator("myIter1"); assert iter.hasNext() : "unable to read event"; DbusEvent event1 = iter.next(); assert iter.hasNext() : "unable to read event"; DbusEvent event2 = iter.next(); assert iter.hasNext() : "unable to read event"; DbusEvent event3 = iter.next(); initMockStreamConsumer3OptEventFullLifecycle(mockConsumer1, event1, event2, event3, keyCounts); initMockStreamConsumer3OptEventFullLifecycle(mockConsumer2, event1, event2, event3, keyCounts); initMockStreamConsumer3OptEventFullLifecycle(mockConsumer3, event1, event2, event3, keyCounts); assert3EventFullLifecycle(callback, event1, event2, event3); EasyMock.verify(mockConsumer1); EasyMock.verify(mockConsumer2); EasyMock.verify(mockConsumer3); assert (keyCounts.get(1L).get() + keyCounts.get(2L).get() + keyCounts.get(3L).get()) == 3 : "invalid number of calls: " + keyCounts.get(1L).get() + "," + keyCounts.get(2L).get() + "," + keyCounts.get(3L).get(); }
@Test(groups = {"small", "functional"}) public void test3GroupedStreamConsumersWithFailure() { LOG.info("\n\nstarting test3GroupedStreamConsumersWithFailure()"); Hashtable<Long, AtomicInteger> keyCounts = new Hashtable<Long, AtomicInteger>(); DbusEventBuffer eventsBuf = new DbusEventBuffer(_generic100KBufferStaticConfig); eventsBuf.start(0); eventsBuf.startEvents(); initBufferWithEvents(eventsBuf, 1, 1, (short)1, keyCounts); initBufferWithEvents(eventsBuf, 2, 2, (short)3, keyCounts); eventsBuf.endEvents(100L); DatabusStreamConsumer mockConsumer1 = EasyMock.createStrictMock("consumer1", DatabusStreamConsumer.class); EasyMock.makeThreadSafe(mockConsumer1, true); DatabusStreamConsumer mockConsumer2 = EasyMock.createStrictMock("consumer2", DatabusStreamConsumer.class); EasyMock.makeThreadSafe(mockConsumer2, true); DatabusStreamConsumer mockConsumer3 = EasyMock.createStrictMock("consumer3", DatabusStreamConsumer.class); EasyMock.makeThreadSafe(mockConsumer3, true); DatabusCombinedConsumer sdccMockConsumer1 = new SelectingDatabusCombinedConsumer(mockConsumer1); DatabusCombinedConsumer sdccMockConsumer2 = new SelectingDatabusCombinedConsumer(mockConsumer2); DatabusCombinedConsumer sdccMockConsumer3 = new SelectingDatabusCombinedConsumer(mockConsumer3); List<String> sources = new ArrayList<String>(); Map<Long, IdNamePair> sourcesMap = new HashMap<Long, IdNamePair>(); for (int i = 1; i <= 3; ++i) { IdNamePair sourcePair = new IdNamePair((long)i, "source" + i); sources.add(sourcePair.getName()); sourcesMap.put(sourcePair.getId(), sourcePair); } DatabusV2ConsumerRegistration consumerReg1 = new DatabusV2ConsumerRegistration( Arrays.asList(sdccMockConsumer1, sdccMockConsumer2, sdccMockConsumer3), sources, null); List<DatabusV2ConsumerRegistration> allRegistrations = Arrays.asList(consumerReg1); MultiConsumerCallback callback = new MultiConsumerCallback( allRegistrations, Executors.newCachedThreadPool(), 1000, new StreamConsumerCallbackFactory(null, null), null, null, null, null); callback.setSourceMap(sourcesMap); DbusEventBuffer.DbusEventIterator iter = eventsBuf.acquireIterator("myIter1"); assert iter.hasNext() : "unable to read event"; DbusEvent event1 = iter.next(); assert iter.hasNext() : "unable to read event"; DbusEvent event2 = iter.next(); assert iter.hasNext() : "unable to read event"; DbusEvent event3 = iter.next(); initMockFailingStreamConsumer3OptEventFullLifecycle(mockConsumer1, event1, event2, event3, keyCounts); initMockFailingStreamConsumer3OptEventFullLifecycle(mockConsumer2, event1, event2, event3, keyCounts); initMockFailingStreamConsumer3OptEventFullLifecycle(mockConsumer3, event1, event2, event3, keyCounts); assert3EventFullLifecycleWithFailure(callback, event1, event2, event3); EasyMock.verify(mockConsumer1); EasyMock.verify(mockConsumer2); EasyMock.verify(mockConsumer3); assert (keyCounts.get(1L).get() + keyCounts.get(2L).get() + keyCounts.get(3L).get()) == 3 : "invalid number of calls: " + keyCounts.get(1L).get() + "," + keyCounts.get(2L).get() + "," + keyCounts.get(3L).get(); }