@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();
}