/*
    * Licensed to the Apache Software Foundation (ASF) under one
    * or more contributor license agreements.  See the NOTICE file
    * distributed with this work for additional information
    * regarding copyright ownership.  The ASF licenses this file
    * to you 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.
   */
  
  package org.apache.giraph.io.hbase;
  
  import java.io.IOException;
  import org.apache.giraph.io.VertexOutputFormat;
  import org.apache.giraph.io.VertexWriter;
  import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
  import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
  import org.apache.hadoop.io.Writable;
  import org.apache.hadoop.io.WritableComparable;
  import org.apache.hadoop.mapreduce.JobContext;
  import org.apache.hadoop.mapreduce.OutputCommitter;
  import org.apache.hadoop.mapreduce.RecordWriter;
  import org.apache.hadoop.mapreduce.TaskAttemptContext;
  
  /**
   *
   * Base class for writing Vertex mutations back to specific
   * rows in an HBase table. This class wraps an instance of TableOutputFormat
   * for easy configuration with the existing properties.
   *
   * Setting conf.set(TableOutputFormat.OUTPUT_TABLE, "out_table");
   * will properly delegate to the TableOutputFormat instance contained
   * in this class. The Configurable interface prevents specific
   * wrapper methods from having to be called.
   *
   * Works with {@link HBaseVertexInputFormat}
   *
   * @param <I> Vertex index value
   * @param <V> Vertex value
   * @param <E> Edge value
   */
  @SuppressWarnings("rawtypes")
  public abstract class HBaseVertexOutputFormat<
          I extends WritableComparable,
          V extends Writable,
          E extends Writable>
          extends VertexOutputFormat
                  <I, V, E> {
  
    /**
     * delegate output format that writes to HBase
     */
    protected static final TableOutputFormat<ImmutableBytesWritable>
    BASE_FORMAT = new TableOutputFormat<ImmutableBytesWritable>();
  
    /**
     *   Constructor
     *
     *   Simple class which takes an instance of RecordWriter
     *   over Writable objects. Subclasses are
     *   expected to implement writeVertex()
     *
     * @param <I> Vertex index value
     * @param <V> Vertex value
     * @param <E> Edge value
     */
    public abstract static class HBaseVertexWriter<
        I extends WritableComparable,
        V extends Writable,
        E extends Writable>
        extends VertexWriter<I, V, E> {
  
      /** Context */
      private TaskAttemptContext context;
      /** Record writer instance */
      private RecordWriter<ImmutableBytesWritable, Writable> recordWriter;
  
     /**
      * Sets up base table output format and creates a record writer.
      * @param context task attempt context
      */
      public HBaseVertexWriter(TaskAttemptContext context)
        throws IOException, InterruptedException {
        BASE_FORMAT.setConf(context.getConfiguration());
        this.recordWriter = BASE_FORMAT.getRecordWriter(context);
      }
  
      /**
       * initialize
       *
       * @param context Context used to write the vertices.
      * @throws IOException
      */
     public void initialize(TaskAttemptContext context)
       throws IOException {
       this.context = context;
     }
 
     /**
      * close
      *
      * @param context the context of the task
      * @throws IOException
      * @throws InterruptedException
      */
     public void close(TaskAttemptContext context)
       throws IOException, InterruptedException {
       recordWriter.close(context);
     }
 
     /**
      * Get the table record writer;
      *
      * @return Record writer to be used for writing.
      */
     public RecordWriter<ImmutableBytesWritable,
             Writable> getRecordWriter() {
       return recordWriter;
     }
 
     /**
      * getContext
      *
      * @return Context passed to initialize.
      */
     public TaskAttemptContext getContext() {
       return context;
     }
   }
 
   /**
    * checkOutputSpecs
    *
    * @param context information about the job
    * @throws IOException
    * @throws InterruptedException
    */
   public void checkOutputSpecs(JobContext context)
     throws IOException, InterruptedException {
     BASE_FORMAT.checkOutputSpecs(context);
   }
 
   /**
    * getOutputCommitter
    *
    * @param context the task context
    * @return  OutputCommitter ouputCommitter
    * @throws IOException
    * @throws InterruptedException
    */
   public OutputCommitter getOutputCommitter(
     TaskAttemptContext context)
     throws IOException, InterruptedException {
     BASE_FORMAT.setConf(getConf());
     return BASE_FORMAT.getOutputCommitter(context);
   }
 }