/**
 * Copyright (c) 2011 Yahoo! Inc. All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License. See accompanying LICENSE file.
 */
package com.yahoo.omid.notifications;

import java.io.IOException;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;

public class TransactionCommittedRegionCoprocessor extends BaseRegionObserver {

    private static final Logger logger = Logger.getLogger(TransactionCommittedRegionCoprocessor.class);

    @Override
    public void start(CoprocessorEnvironment e) throws IOException {
        logger.trace("TransactionCommitterRegionObserver started");
    }

    @Override
    public void prePut(ObserverContext<RegionCoprocessorEnvironment> c, Put put, WALEdit edit, boolean writeToWAL)
            throws IOException {

        byte[] tableNameAsBytes = c.getEnvironment().getRegion().getTableDesc().getName();
        String tableName = Bytes.toString(tableNameAsBytes);

        if (tableName.equals(".META.")|| tableName.equals("-ROOT-")) {
        	// do not instrument hbase meta tables
        	// TODO allow filtering out more tables?
        	return;
        }

        //logger.trace("We're in pre-put");
        HTable table = new HTable(HBaseConfiguration.create(), tableName);
        
        Map<byte[], List<KeyValue>> kvs = put.getFamilyMap();

        for (List<KeyValue> kvl : kvs.values()) {
            for (KeyValue kv : kvl) {
                String cf = Bytes.toString(kv.getFamily());
                // TODO Here we should filter the columns that are not requested to be observed, 
                // but cannot access to the required structures from a BaseRegionObserver
                if(!cf.equals(Constants.HBASE_META_CF)) {
                    String col = Bytes.toString(kv.getQualifier());
                    // Pattern for notify column in framework's metadata column family: <cf>/<c>-notify
                    put.add(Bytes.toBytes(Constants.HBASE_META_CF),
                            Bytes.toBytes(cf + "/" + col + Constants.HBASE_NOTIFY_SUFFIX), kv.getTimestamp(), Bytes.toBytes("true"));
                    logger.trace("This is the put added : " + put);
                }
            }
        }
        table.close();
        //logger.trace("Out of pre-put");
    }

}

@@ -848,112 +848,15 @@
 ase.
-HBaseConfiguration;%0Aimport org.apache.hadoop.hbase.KeyValue;%0Aimport org.apache.hadoop.hbase.client.HTabl
+KeyValu
 e;%0Ai
@@ -1208,18 +1208,36 @@
 org.
-apache.log
+slf4j.Logger;%0Aimport org.slf
 4j.L
@@ -1241,16 +1241,23 @@
 j.Logger
+Factory
 ;%0A%0Apubli
@@ -1379,16 +1379,23 @@
 = Logger
+Factory
 .getLogg
@@ -1956,16 +1956,17 @@
 .META.%22)
+ 
 %7C%7C table
@@ -1998,17 +1998,20 @@
         
+    
-%09
 // do no
@@ -2049,17 +2049,20 @@
         
-%09
+    
 // TODO 
@@ -2106,145 +2106,18 @@
     
-%09return;%0A        %7D%0A%0A        //logger.trace(%22We're in pre-put%22);%0A        HTable table = new HTable(HBaseConfiguration.create(), tableName)
+    return
 ;%0A  
@@ -2122,16 +2122,18 @@
         
+%7D%0A
 %0A       
@@ -2432,17 +2432,16 @@
 bserved,
- 
 %0A       
@@ -2541,16 +2541,17 @@
       if
+ 
 (!cf.equ
@@ -2935,207 +2935,102 @@
 p(),
- Bytes.toBytes(%22true%22));%0A                    logger.trace(%22This is the put added : %22 + put);%0A                %7D%0A            %7D%0A        %7D%0A        table.close();%0A        //logger.trace(%22Out of pre-put%22);
+%0A                            Bytes.toBytes(%22true%22));%0A                %7D%0A            %7D%0A        %7D
 %0A   

Remove instantiation of HTable in coprocessor - not used + harms performance