@Test
public void testAddEventToBuffer() throws InvalidConfigException,
    UnsupportedKeyException, DatabusException
{

  // No rate control
  long rate = 0;
  PhysicalSourceStaticConfig pssc = buildPssc(rate, 0L);
  long scn = 10;
  DbusEventBuffer mb = (DbusEventBuffer)createBufMult(pssc);
  GoldenGateEventProducer gg = new GoldenGateEventProducer(pssc, null, mb,
      null, null);
  List<TransactionState.PerSourceTransactionalUpdate> dbUpdates = new ArrayList<TransactionState.PerSourceTransactionalUpdate>(
      10);
  int sourceId = 505;
  HashSet<DBUpdateImage> db = new HashSet<DBUpdateImage>();
  Object key = new String("name");
  Schema.Type keyType = Schema.Type.RECORD;
  ColumnsState.KeyPair kp = new ColumnsState.KeyPair(key, keyType);
  ArrayList<ColumnsState.KeyPair> keyPairs = new ArrayList<ColumnsState.KeyPair>(
      1);
  keyPairs.add(kp);
  Schema s = Schema.parse(avroSchema);
  GenericRecord gr = new GenericData.Record(s);
  gr.put("name", "phani");
  DBUpdateImage dbi = new DBUpdateImage(keyPairs, scn, gr, s,
      DbUpdateState.DBUpdateImage.OpType.INSERT,false);
  db.add(dbi);
  TransactionState.PerSourceTransactionalUpdate dbUpdate = new TransactionState.PerSourceTransactionalUpdate(
      sourceId, db);
  dbUpdates.add(dbUpdate);
  long timestamp = System.nanoTime();
  gg.addEventToBuffer(dbUpdates, new TransactionInfo(0, 0, timestamp, scn));
  Assert.assertEquals(gg.getRateControl().getNumSleeps(), 0);
  DbusEventIterator iter  = mb.acquireIterator("test");
  int count = 0;
  long eventTs = 0;
  while(iter.hasNext()) {
    DbusEvent e = iter.next();
    if(count==1) { // first event prev control event
      eventTs = e.timestampInNanos();
    }
    count ++;
  }
   Assert.assertEquals("Event timestamp in Ns", timestamp, eventTs);
  Assert.assertEquals("Got events " , 3, count);
  return;
}