@Test
public void testDbUpdatesMergeUtils()
  throws Exception
{

  short[] sourceIds = new short[] { 505, 506 };
  long scn = 10;
  // Case when 2 txn have same source and keys
  {
    String key = "key1";
    String oldVal = "val1";
    Map<String, String> keyVals1 = new HashMap<String, String>();
    keyVals1.put(key, oldVal);
    Map<String, String> keyVals2 = new HashMap<String, String>();
    String newVal = "val2";
    keyVals2.put(key, newVal);
    List<TransactionState.PerSourceTransactionalUpdate> oldDbUpdates =
        generateUpdatesForSchema2(sourceIds, keyVals1, scn);
    List<TransactionState.PerSourceTransactionalUpdate> newDbUpdates =
        generateUpdatesForSchema2(sourceIds, keyVals2, scn);
    List<TransactionState.PerSourceTransactionalUpdate> result =
        DBUpdatesMergeUtils.mergeTransactionData(newDbUpdates, oldDbUpdates);
    Assert.assertEquals("Number of Source batches", 2, result.size());
    Assert.assertEquals("Result Set Size", result.get(0).getDbUpdatesSet().size(), 1);
    Assert.assertEquals("Result Set Size", result.get(1).getDbUpdatesSet().size(), 1);
    DBUpdateImage dbi = result.get(0).getDbUpdatesSet().iterator().next();
    DBUpdateImage expDbi = newDbUpdates.get(0).getDbUpdatesSet().iterator().next();
    Assert.assertEquals("SCN Check", scn, dbi.getScn());
    String k = (String) dbi.getGenericRecord().get("name1");
    String v = (String) dbi.getGenericRecord().get("name2");
    String expK = (String) expDbi.getGenericRecord().get("name1");
    String expV = (String) expDbi.getGenericRecord().get("name2");
    Assert.assertEquals("Key Check", expK, k);
    Assert.assertEquals("Val Check", expV, v);
    dbi = result.get(1).getDbUpdatesSet().iterator().next();
    expDbi = newDbUpdates.get(1).getDbUpdatesSet().iterator().next();
    Assert.assertEquals("SCN Check", scn, dbi.getScn());
    k = (String) dbi.getGenericRecord().get("name1");
    v = (String) dbi.getGenericRecord().get("name2");
    expK = (String) expDbi.getGenericRecord().get("name1");
    expV = (String) expDbi.getGenericRecord().get("name2");
    Assert.assertEquals("Key Check", expK, k);
    Assert.assertEquals("Val Check", expV, v);
  }
  // Case when 2 txn have same source but different keys
  {
    String key = "key1";
    String oldVal = "val1";
    Map<String, String> keyVals1 = new HashMap<String, String>();
    keyVals1.put(key, oldVal);
    Map<String, String> keyVals2 = new HashMap<String, String>();
    key = "key2";
    String newVal = "val2";
    keyVals2.put(key, newVal);
    List<TransactionState.PerSourceTransactionalUpdate> oldDbUpdates =
        generateUpdatesForSchema2(sourceIds, keyVals1, scn);
    List<TransactionState.PerSourceTransactionalUpdate> newDbUpdates =
        generateUpdatesForSchema2(sourceIds, keyVals2, scn);
    List<TransactionState.PerSourceTransactionalUpdate> result =
        DBUpdatesMergeUtils.mergeTransactionData(newDbUpdates, oldDbUpdates);
    Assert.assertEquals("Number of Source batches", 2, result.size());
    Assert.assertEquals("Result Set Size", result.get(0).getDbUpdatesSet().size(), 2);
    Assert.assertEquals("Result Set Size", result.get(1).getDbUpdatesSet().size(), 2);
    Iterator<DBUpdateImage> itr = result.get(0).getDbUpdatesSet().iterator();
    DBUpdateImage dbi = itr.next();
    Assert.assertEquals("SCN Check", scn, dbi.getScn());
    String k = (String) dbi.getGenericRecord().get("name1");
    String v = (String) dbi.getGenericRecord().get("name2");
    dbi = itr.next();
    Assert.assertEquals("SCN Check", scn, dbi.getScn());
    String k2 = (String) dbi.getGenericRecord().get("name1");
    String v2 = (String) dbi.getGenericRecord().get("name2");
    if (keyVals1.containsKey(k))
    {
      Assert.assertEquals("Val Check", keyVals1.get(k), v);
      Assert.assertEquals("Val Check 2", keyVals2.get(k2), v2);
    }
    else
    {
      Assert.assertEquals("Val Check", keyVals2.get(k), v);
      Assert.assertEquals("Val Check 2", keyVals1.get(k2), v2);
    }
    itr = result.get(1).getDbUpdatesSet().iterator();
    dbi = itr.next();
    Assert.assertEquals("SCN Check", scn, dbi.getScn());
    k = (String) dbi.getGenericRecord().get("name1");
    v = (String) dbi.getGenericRecord().get("name2");
    dbi = itr.next();
    Assert.assertEquals("SCN Check", scn, dbi.getScn());
    k2 = (String) dbi.getGenericRecord().get("name1");
    v2 = (String) dbi.getGenericRecord().get("name2");
    if (keyVals1.containsKey(k))
    {
      Assert.assertEquals("Val Check", keyVals1.get(k), v);
      Assert.assertEquals("Val Check 2", keyVals2.get(k2), v2);
    }
    else
    {
      Assert.assertEquals("Val Check", keyVals2.get(k), v);
      Assert.assertEquals("Val Check 2", keyVals1.get(k2), v2);
    }
  }
  // Case when 2 txns have different sources
  {
    String key = "key1";
    String oldVal = "val1";
    Map<String, String> keyVals1 = new HashMap<String, String>();
    keyVals1.put(key, oldVal);
    Map<String, String> keyVals2 = new HashMap<String, String>();
    key = "key2";
    String newVal = "val2";
    keyVals2.put(key, newVal);
    List<TransactionState.PerSourceTransactionalUpdate> oldDbUpdates =
        generateUpdatesForSchema2(new short[] { sourceIds[0] }, keyVals1, scn);
    List<TransactionState.PerSourceTransactionalUpdate> newDbUpdates =
        generateUpdatesForSchema2(new short[] { sourceIds[1] }, keyVals2, scn);
    List<TransactionState.PerSourceTransactionalUpdate> result =
        DBUpdatesMergeUtils.mergeTransactionData(newDbUpdates, oldDbUpdates);
    Assert.assertEquals("Number of Source batches", 2, result.size());
    Assert.assertEquals("Result Set Size", result.get(0).getDbUpdatesSet().size(), 1);
    Assert.assertEquals("Result Set Size", result.get(1).getDbUpdatesSet().size(), 1);
    Iterator<DBUpdateImage> itr = result.get(0).getDbUpdatesSet().iterator();
    DBUpdateImage dbi = itr.next();
    Assert.assertEquals("SCN Check", scn, dbi.getScn());
    String k = (String) dbi.getGenericRecord().get("name1");
    String v = (String) dbi.getGenericRecord().get("name2");
    Assert.assertEquals("Val Check", keyVals1.get(k), v);
    itr = result.get(1).getDbUpdatesSet().iterator();
    dbi = itr.next();
    Assert.assertEquals("SCN Check", scn, dbi.getScn());
    k = (String) dbi.getGenericRecord().get("name1");
    v = (String) dbi.getGenericRecord().get("name2");
    Assert.assertEquals("Val Check", keyVals2.get(k), v);
  }
}