package org.opencb.opencga.storage.hadoop.utils;

import org.apache.commons.collections.CollectionUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.KeyValueUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.opencb.commons.io.DataWriter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.*;

/**
 * Created on 31/01/18.
 *
 * @author Jacobo Coll &lt;jacobo167@gmail.com&gt;
 */
public abstract class AbstractHBaseDataWriter<T, M extends Mutation> implements DataWriter<T> {

    protected final HBaseManager hBaseManager;
    protected final String tableName;
    private BufferedMutator mutator;
    private final Logger logger = LoggerFactory.getLogger(AbstractHBaseDataWriter.class);

    public AbstractHBaseDataWriter(HBaseManager hBaseManager, String tableName) {
        this.hBaseManager = new HBaseManager(Objects.requireNonNull(hBaseManager));
        this.tableName = tableName;
    }

    protected abstract List<M> convert(List<T> batch);

    @Override
    public boolean open() {
        try {
            mutator = hBaseManager.getConnection().getBufferedMutator(TableName.valueOf(tableName));
        } catch (IOException e) {
            throw new UncheckedIOException("Failed to connect to Hbase", e);
        }
        return true;
    }

    @Override
    public boolean write(List<T> list) {
        try {
            mutate(convert(list));
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
        return true;
    }

    protected final synchronized void mutate(List<M> mutations) throws IOException {
        try {
            mutator.mutate(mutations);
        } catch (IllegalArgumentException e) {
            // Try to extend the information regarding the InvalidArgumentException, in case of being a PUT validation exception
            if (CollectionUtils.isNotEmpty(mutations) && mutations.get(0) instanceof Put && mutator instanceof BufferedMutatorImpl) {
                try {
                    hBaseManager.act(tableName, table -> {
                        for (M mutation : mutations) {
                            if (mutation instanceof Put) {
                                try {
                                    int maxKeyValueSize = hBaseManager.getConf().getInt(
                                            ConnectionConfiguration.MAX_KEYVALUE_SIZE_KEY,
                                            ConnectionConfiguration.MAX_KEYVALUE_SIZE_DEFAULT);
                                    validatePut((Put) mutation, maxKeyValueSize);
                                } catch (RuntimeException e1) {
                                    // Don't print the whole stacktrace
                                    logger.error("Invalid put operation on RowKey '" + Bytes.toStringBinary(mutation.getRow()) + "', "
                                            + e1.getMessage());
                                }
                            }
                        }
                    });
                } catch (IOException ioe) {
                    // Do not propagate this IOException.
                    logger.error("Unexpected IOException", e);
                }
            }
            throw e;
        }
    }

    // validate for well-formedness
    public static void validatePut(Put put, int maxKeyValueSize) throws IllegalArgumentException {
        if (put.isEmpty()) {
            throw new IllegalArgumentException("No columns to insert");
        }
        if (maxKeyValueSize > 0) {
            for (List<Cell> list : put.getFamilyCellMap().values()) {
                for (Cell cell : list) {
                    if (KeyValueUtil.length(
                            cell.getRowLength(),
                            cell.getFamilyLength(),
                            cell.getQualifierLength(),
                            cell.getValueLength(),
                            cell.getTagsLength(), true) > maxKeyValueSize) {
                        throw new IllegalArgumentException("KeyValue size too large");
                    }
                }
            }
        }
    }

    public final synchronized void flush() {
        try {
            mutator.flush();
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    }

    @Override
    public boolean post() {
        flush();
        return true;
    }

    @Override
    public boolean close() {
        try {
            mutator.close();
            hBaseManager.close();
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
        return true;
    }
}

@@ -139,53 +139,8 @@
 ll;%0A
-import org.apache.hadoop.hbase.KeyValueUtil;%0A
 impo
@@ -442,17 +442,46 @@
 va.util.
-*
+List;%0Aimport java.util.Objects
 ;%0A%0A/**%0A 
@@ -3804,32 +3804,29 @@
     
-if (KeyValueUtil.length(
+long length = 8L + 12
 %0A   
@@ -3841,32 +3841,34 @@
                 
+ +
  cell.getRowLeng
@@ -3863,33 +3863,32 @@
 l.getRowLength()
-,
 %0A               
@@ -3891,32 +3891,34 @@
                 
+ +
  cell.getFamilyL
@@ -3916,33 +3916,32 @@
 etFamilyLength()
-,
 %0A               
@@ -3944,32 +3944,34 @@
                 
+ +
  cell.getQualifi
@@ -3972,33 +3972,32 @@
 ualifierLength()
-,
 %0A               
@@ -4000,32 +4000,34 @@
                 
+ +
  cell.getValueLe
@@ -4032,17 +4032,17 @@
 Length()
-,
+;
 %0A       
@@ -4058,43 +4058,18 @@
     
-        cell.getTagsLength(), true)
+if (length
  %3E m

Fix compilation issue with hdp2.6 #1695