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