HBASE-28268 Provide option to skip wal while using TableOutputFormat

hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/TableOutputFormat.java

@@ -33,6 +33,7 @@
 import org.apache.hadoop.hbase.client.Connection;
 import org.apache.hadoop.hbase.client.ConnectionFactory;
 import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Durability;
 import org.apache.hadoop.hbase.client.Mutation;
 import org.apache.hadoop.hbase.client.Put;
 import org.apache.hadoop.mapreduce.JobContext;
@@ -56,6 +57,15 @@ public class TableOutputFormat<KEY> extends OutputFormat<KEY, Mutation> implemen
   /** Job parameter that specifies the output table. */
   public static final String OUTPUT_TABLE = "hbase.mapred.outputtable";
 
+  /** Property value to use write-ahead logging */
+  public static final boolean WAL_ON = true;
+
+  /** Property value to disable write-ahead logging */
+  public static final boolean WAL_OFF = false;
+
+  /** Set this to {@link #WAL_OFF} to turn off write-ahead logging (WAL) */
+  public static final String WAL_PROPERTY = "hbase.mapreduce.tableoutputformat.write.wal";
+
   /**
    * Optional job parameter to specify a peer cluster. Used specifying remote cluster when copying
    * between hbase clusters (the source is picked up from <code>hbase-site.xml</code>).
@@ -139,12 +149,14 @@ protected class TableRecordWriter extends RecordWriter<KEY, Mutation> {
 
     private Connection connection;
     private BufferedMutator mutator;
+    boolean useWriteAheadLogging;
 
     public TableRecordWriter() throws IOException {
       this.connection = createConnection(conf);
       String tableName = conf.get(OUTPUT_TABLE);
       this.mutator = connection.getBufferedMutator(TableName.valueOf(tableName));
       LOG.info("Created table instance for " + tableName);
+      this.useWriteAheadLogging = conf.getBoolean(WAL_PROPERTY, WAL_ON);
     }
 
     /**
@@ -178,6 +190,9 @@ public void write(KEY key, Mutation value) throws IOException {
       if (!(value instanceof Put) && !(value instanceof Delete)) {
         throw new IOException("Pass a Delete or a Put");
       }
+      if (!useWriteAheadLogging) {
+        value.setDurability(Durability.SKIP_WAL);
+      }
       mutator.mutate(value);
     }
   }

hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestTableOutputFormat.java

@@ -0,0 +1,130 @@
+/*
+ * 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.hadoop.hbase.mapreduce;
+
+import java.io.IOException;
+import javax.validation.constraints.Null;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseClassTestRule;
+import org.apache.hadoop.hbase.HBaseTestingUtil;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Durability;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.testclassification.MediumTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.mapreduce.RecordWriter;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.mockito.Mockito;
+
+/**
+ * Simple Tests to check whether the durability of the Mutation is changed or not, for
+ * {@link TableOutputFormat} if {@link TableOutputFormat#WAL_PROPERTY} is set to false.
+ */
+@Category(MediumTests.class)
+public class TestTableOutputFormat {
+  @ClassRule
+  public static final HBaseClassTestRule CLASS_RULE =
+    HBaseClassTestRule.forClass(TestTableOutputFormat.class);
+
+  private static final HBaseTestingUtil util = new HBaseTestingUtil();
+  private static final TableName TABLE_NAME = TableName.valueOf("TEST_TABLE");
+  private static final byte[] columnFamily = Bytes.toBytes("f");
+  private static Configuration conf;
+  private static RecordWriter<Null, Mutation> writer;
+  private static TaskAttemptContext context;
+  private static TableOutputFormat<Null> tableOutputFormat;
+
+  @BeforeClass
+  public static void setUp() throws Exception {
+    util.startMiniCluster();
+    util.createTable(TABLE_NAME, columnFamily);
+
+    conf = new Configuration(util.getConfiguration());
+    context = Mockito.mock(TaskAttemptContext.class);
+    tableOutputFormat = new TableOutputFormat<>();
+    conf.set(TableOutputFormat.OUTPUT_TABLE, "TEST_TABLE");
+  }
+
+  @AfterClass
+  public static void tearDown() throws Exception {
+    util.shutdownMiniCluster();
+  }
+
+  @After
+  public void close() throws IOException, InterruptedException {
+    if (writer != null && context != null) {
+      writer.close(context);
+    }
+  }
+
+  @Test
+  public void testTableOutputFormatWhenWalIsOFFForPut() throws IOException, InterruptedException {
+    // setting up the configuration for the TableOutputFormat, with writing to the WAL off.
+    conf.setBoolean(TableOutputFormat.WAL_PROPERTY, TableOutputFormat.WAL_OFF);
+    tableOutputFormat.setConf(conf);
+
+    writer = tableOutputFormat.getRecordWriter(context);
+
+    // creating mutation of the type put
+    Put put = new Put("row1".getBytes());
+    put.addColumn(columnFamily, Bytes.toBytes("aa"), Bytes.toBytes("value"));
+
+    // verifying whether durability of mutation is USE_DEFAULT or not, before commiting write.
+    Assert.assertEquals("Durability of the mutation should be USE_DEFAULT", Durability.USE_DEFAULT,
+      put.getDurability());
+
+    writer.write(null, put);
+
+    // verifying whether durability of mutation got changed to the SKIP_WAL or not.
+    Assert.assertEquals("Durability of the mutation should be SKIP_WAL", Durability.SKIP_WAL,
+      put.getDurability());
+  }
+
+  @Test
+  public void testTableOutputFormatWhenWalIsOFFForDelete()
+    throws IOException, InterruptedException {
+    // setting up the configuration for the TableOutputFormat, with writing to the WAL off.
+    conf.setBoolean(TableOutputFormat.WAL_PROPERTY, TableOutputFormat.WAL_OFF);
+    tableOutputFormat.setConf(conf);
+
+    writer = tableOutputFormat.getRecordWriter(context);
+
+    // creating mutation of the type delete
+    Delete delete = new Delete("row2".getBytes());
+    delete.addColumn(columnFamily, Bytes.toBytes("aa"));
+
+    // verifying whether durability of mutation is USE_DEFAULT or not, before commiting write.
+    Assert.assertEquals("Durability of the mutation should be USE_DEFAULT", Durability.USE_DEFAULT,
+      delete.getDurability());
+
+    writer.write(null, delete);
+
+    // verifying whether durability of mutation got changed from USE_DEFAULT to the SKIP_WAL or not.
+    Assert.assertEquals("Durability of the mutation should be SKIP_WAL", Durability.SKIP_WAL,
+      delete.getDurability());
+  }
+}