package org.apache.flume.sink.hbase;

import static org.junit.Assert.*;

import org.hbase.async.PutRequest;
import org.junit.Test;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.flume.Event;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.Context;

public class TestSimpleAsyncHbaseEventSerializer {
   @Test
   public void testSetEvent() {
      // Event
      byte[] body = new String("BODY123").getBytes();
      Map <String, String> headers = new HashMap<String, String>();
      for (int icnt=0;icnt<5;icnt++) {
         headers.put("h"+icnt, "v"+icnt);
      }
      headers.put("host", "10.144.123.19");
      Event event = EventBuilder.withBody(body, headers);
      assertEquals("HEADER1","v1",event.getHeaders().get("h1"));
      assertEquals("BODY","BODY123",new String(event.getBody()));
      // Context
      Context cotx = new Context();
      cotx.put("payloadColumn", "evtBody");
      cotx.put("incrementColumn", "iCol");
      cotx.put("rowPrefix", "evKey");
      cotx.put("suffix", "timestamp");
      cotx.put("incrementRow", "iRow");      
      // Serializer
      SimpleAsyncHbaseEventSerializer ser = new SimpleAsyncHbaseEventSerializer();
      ser.initialize("t1".getBytes(), "fam1".getBytes());
      ser.configure(cotx);
      ser.setEvent(event);
      List<PutRequest> actions = ser.getActions();
      for(PutRequest req:actions){
         System.out.println(new String("T=["+new String(req.table())
         +"],K=["+new String(req.key())
         +"],F=["+new String(req.family())
         +"],Q=["+new String(req.qualifier())
         +"],V=["+new String(req.value())+"]"));
      }
      
   }

}

@@ -653,21 +653,15 @@
 %22, %22
-10.144.123.19
+srchost
 %22);%0A
@@ -1023,16 +1023,17 @@
 , %22evKey
+-
 %22);%0A    
@@ -1301,16 +1301,35 @@
 (cotx);%0A
+      // 1st event%0A
       se
@@ -1688,22 +1688,385 @@
       %7D%0A
-
       
+// 2nd event%0A      ser.setEvent(event);%0A      actions = ser.getActions();%0A      for(PutRequest req:actions)%7B%0A         System.out.println(new String(%22T=%5B%22+new String(req.table())%0A         +%22%5D,K=%5B%22+new String(req.key())%0A         +%22%5D,F=%5B%22+new String(req.family())%0A         +%22%5D,Q=%5B%22+new String(req.qualifier())%0A         +%22%5D,V=%5B%22+new String(req.value())+%22%5D%22));%0A      %7D
 %0A   %7D%0A%0A%7D

Update TestSimpleAsyncHbaseEventSerializer.java