Only use ExtendedCellSerialization for WALPlayer

Author: bbeaudreault

hbase-backup/pom.xml

@@ -29,7 +29,6 @@
   <artifactId>hbase-backup</artifactId>
   <name>Apache HBase - Backup</name>
   <description>Backup for HBase</description>
-  <!-- trigger build -->
   <dependencies>
     <!-- Intra-project dependencies -->
     <dependency>

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

@@ -30,6 +30,7 @@
 import java.nio.charset.Charset;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
@@ -159,6 +160,15 @@ protected static byte[] combineTableNameSuffix(byte[] tableName, byte[] suffix)
   static final String MULTI_TABLE_HFILEOUTPUTFORMAT_CONF_KEY =
     "hbase.mapreduce.use.multi.table.hfileoutputformat";
 
+  /**
+   * ExtendedCell and ExtendedCellSerialization are InterfaceAudience.Private. We expose this config
+   * package-private for internal usage for jobs like WALPlayer which need to use features of
+   * ExtendedCell.
+   */
+  static final String EXTENDED_CELL_SERIALIZATION_ENABLED_KEY =
+    "hbase.mapreduce.hfileoutputformat.extendedcell.enabled";
+  static final boolean EXTENDED_CELL_SERIALIZATION_ENABLED_DEFULT = false;
+
   public static final String REMOTE_CLUSTER_CONF_PREFIX = "hbase.hfileoutputformat.remote.cluster.";
   public static final String REMOTE_CLUSTER_ZOOKEEPER_QUORUM_CONF_KEY =
     REMOTE_CLUSTER_CONF_PREFIX + "zookeeper.quorum";
@@ -619,13 +629,12 @@ static void configureIncrementalLoad(Job job, List<TableInfo> multiTableInfo,
       LOG.warn("Unknown map output value type:" + job.getMapOutputValueClass());
     }
 
-    // Order matters here. Hadoop's SerializationFactory runs through serializations in the order
-    // they are registered. Register ExtendedCellSerialization before CellSerialization because both
-    // work for ExtendedCells but ExtendedCellSerialization handles them properly.
     conf.setStrings("io.serializations", conf.get("io.serializations"),
       MutationSerialization.class.getName(), ResultSerialization.class.getName(),
       ExtendedCellSerialization.class.getName(), CellSerialization.class.getName());
 
+    mergeSerializations(conf);
+
     if (conf.getBoolean(LOCALITY_SENSITIVE_CONF_KEY, DEFAULT_LOCALITY_SENSITIVE)) {
       LOG.info("bulkload locality sensitive enabled");
     }
@@ -673,6 +682,33 @@ static void configureIncrementalLoad(Job job, List<TableInfo> multiTableInfo,
     LOG.info("Incremental output configured for tables: " + StringUtils.join(allTableNames, ","));
   }
 
+  private static void mergeSerializations(Configuration conf) {
+    List<String> serializations = new ArrayList<>();
+
+    // add any existing values that have been set
+    String[] existing = conf.getStrings("io.serializations");
+    if (existing != null) {
+      Collections.addAll(serializations, existing);
+    }
+
+    serializations.add(MutationSerialization.class.getName());
+    serializations.add(ResultSerialization.class.getName());
+
+    // Add ExtendedCellSerialization, if configured. Order matters here. Hadoop's
+    // SerializationFactory runs through serializations in the order they are registered.
+    // We want to register ExtendedCellSerialization before CellSerialization because both
+    // work for ExtendedCells but only ExtendedCellSerialization handles them properly.
+    if (
+      conf.getBoolean(EXTENDED_CELL_SERIALIZATION_ENABLED_KEY,
+        EXTENDED_CELL_SERIALIZATION_ENABLED_DEFULT)
+    ) {
+      serializations.add(ExtendedCellSerialization.class.getName());
+    }
+    serializations.add(CellSerialization.class.getName());
+
+    conf.setStrings("io.serializations", serializations.toArray(new String[0]));
+  }
+
   public static void configureIncrementalLoadMap(Job job, TableDescriptor tableDescriptor)
     throws IOException {
     Configuration conf = job.getConfiguration();

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

@@ -316,6 +316,10 @@ public Job createSubmittableJob(String[] args) throws IOException {
     if (hfileOutPath != null) {
       LOG.debug("add incremental job :" + hfileOutPath + " from " + inputDirs);
 
+      // WALPlayer needs ExtendedCellSerialization so that sequenceId can be propagated when
+      // sorting cells in CellSortReducer
+      conf.setBoolean(HFileOutputFormat2.EXTENDED_CELL_SERIALIZATION_ENABLED_KEY, true);
+
       // the bulk HFile case
       List<TableName> tableNames = getTableNameList(tables);