[HUDI-7346] Remove usage of org.apache.hadoop.hbase.util.Bytes (#10574)

hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestTableCommand.java

@@ -46,7 +46,6 @@
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.IOException;
-import java.nio.charset.StandardCharsets;
 import java.nio.file.Paths;
 import java.util.Arrays;
 import java.util.Collections;
@@ -55,6 +54,7 @@
 import java.util.Map;
 
 import static org.apache.hudi.common.table.HoodieTableMetaClient.METAFOLDER_NAME;
+import static org.apache.hudi.common.util.StringUtils.fromUTF8Bytes;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -285,6 +285,6 @@ private String getFileContent(String fileToReadStr) throws IOException {
     byte[] data = new byte[(int) fileToRead.length()];
     fis.read(data);
     fis.close();
-    return new String(data, StandardCharsets.UTF_8);
+    return fromUTF8Bytes(data);
   }
 }

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/hbase/SparkHoodieHBaseIndex.java

@@ -61,7 +61,6 @@
 import org.apache.hadoop.hbase.client.Result;
 import org.apache.hadoop.hbase.client.ResultScanner;
 import org.apache.hadoop.hbase.client.Scan;
-import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.spark.Partitioner;
 import org.apache.spark.SparkConf;
@@ -96,6 +95,8 @@
 import static org.apache.hadoop.hbase.security.SecurityConstants.REGIONSERVER_KRB_PRINCIPAL;
 import static org.apache.hadoop.hbase.security.User.HBASE_SECURITY_AUTHORIZATION_CONF_KEY;
 import static org.apache.hadoop.hbase.security.User.HBASE_SECURITY_CONF_KEY;
+import static org.apache.hudi.common.util.StringUtils.fromUTF8Bytes;
+import static org.apache.hudi.common.util.StringUtils.getUTF8Bytes;
 
 /**
  * Hoodie Index implementation backed by HBase.
@@ -107,10 +108,10 @@ public class SparkHoodieHBaseIndex extends HoodieIndex<Object, Object> {
   public static final String DEFAULT_SPARK_DYNAMIC_ALLOCATION_MAX_EXECUTORS_CONFIG_NAME =
       "spark.dynamicAllocation.maxExecutors";
 
-  private static final byte[] SYSTEM_COLUMN_FAMILY = Bytes.toBytes("_s");
-  private static final byte[] COMMIT_TS_COLUMN = Bytes.toBytes("commit_ts");
-  private static final byte[] FILE_NAME_COLUMN = Bytes.toBytes("file_name");
-  private static final byte[] PARTITION_PATH_COLUMN = Bytes.toBytes("partition_path");
+  private static final byte[] SYSTEM_COLUMN_FAMILY = getUTF8Bytes("_s");
+  private static final byte[] COMMIT_TS_COLUMN = getUTF8Bytes("commit_ts");
+  private static final byte[] FILE_NAME_COLUMN = getUTF8Bytes("file_name");
+  private static final byte[] PARTITION_PATH_COLUMN = getUTF8Bytes("partition_path");
 
   private static final Logger LOG = LoggerFactory.getLogger(SparkHoodieHBaseIndex.class);
   private static Connection hbaseConnection = null;
@@ -217,7 +218,7 @@ public void close() {
   }
 
   private Get generateStatement(String key) throws IOException {
-    return new Get(Bytes.toBytes(getHBaseKey(key))).readVersions(1).addColumn(SYSTEM_COLUMN_FAMILY, COMMIT_TS_COLUMN)
+    return new Get(getUTF8Bytes(getHBaseKey(key))).readVersions(1).addColumn(SYSTEM_COLUMN_FAMILY, COMMIT_TS_COLUMN)
         .addColumn(SYSTEM_COLUMN_FAMILY, FILE_NAME_COLUMN).addColumn(SYSTEM_COLUMN_FAMILY, PARTITION_PATH_COLUMN);
   }
 
@@ -272,10 +273,10 @@ private <R> Function2<Integer, Iterator<HoodieRecord<R>>, Iterator<HoodieRecord<
               taggedRecords.add(currentRecord);
               continue;
             }
-            String keyFromResult = Bytes.toString(result.getRow());
-            String commitTs = Bytes.toString(result.getValue(SYSTEM_COLUMN_FAMILY, COMMIT_TS_COLUMN));
-            String fileId = Bytes.toString(result.getValue(SYSTEM_COLUMN_FAMILY, FILE_NAME_COLUMN));
-            String partitionPath = Bytes.toString(result.getValue(SYSTEM_COLUMN_FAMILY, PARTITION_PATH_COLUMN));
+            String keyFromResult = fromUTF8Bytes(result.getRow());
+            String commitTs = fromUTF8Bytes(result.getValue(SYSTEM_COLUMN_FAMILY, COMMIT_TS_COLUMN));
+            String fileId = fromUTF8Bytes(result.getValue(SYSTEM_COLUMN_FAMILY, FILE_NAME_COLUMN));
+            String partitionPath = fromUTF8Bytes(result.getValue(SYSTEM_COLUMN_FAMILY, PARTITION_PATH_COLUMN));
             if (!HoodieIndexUtils.checkIfValidCommit(completedCommitsTimeline, commitTs)) {
               // if commit is invalid, treat this as a new taggedRecord
               taggedRecords.add(currentRecord);
@@ -369,14 +370,14 @@ private Function2<Integer, Iterator<WriteStatus>, Iterator<WriteStatus>> updateL
                     // This is an update, no need to update index
                     continue;
                   }
-                  Put put = new Put(Bytes.toBytes(getHBaseKey(recordDelegate.getRecordKey())));
-                  put.addColumn(SYSTEM_COLUMN_FAMILY, COMMIT_TS_COLUMN, Bytes.toBytes(loc.get().getInstantTime()));
-                  put.addColumn(SYSTEM_COLUMN_FAMILY, FILE_NAME_COLUMN, Bytes.toBytes(loc.get().getFileId()));
-                  put.addColumn(SYSTEM_COLUMN_FAMILY, PARTITION_PATH_COLUMN, Bytes.toBytes(recordDelegate.getPartitionPath()));
+                  Put put = new Put(getUTF8Bytes(getHBaseKey(recordDelegate.getRecordKey())));
+                  put.addColumn(SYSTEM_COLUMN_FAMILY, COMMIT_TS_COLUMN, getUTF8Bytes(loc.get().getInstantTime()));
+                  put.addColumn(SYSTEM_COLUMN_FAMILY, FILE_NAME_COLUMN, getUTF8Bytes(loc.get().getFileId()));
+                  put.addColumn(SYSTEM_COLUMN_FAMILY, PARTITION_PATH_COLUMN, getUTF8Bytes(recordDelegate.getPartitionPath()));
                   mutations.add(put);
                 } else {
                   // Delete existing index for a deleted record
-                  Delete delete = new Delete(Bytes.toBytes(getHBaseKey(recordDelegate.getRecordKey())));
+                  Delete delete = new Delete(getUTF8Bytes(getHBaseKey(recordDelegate.getRecordKey())));
                   mutations.add(delete);
                 }
               }
@@ -616,7 +617,7 @@ public boolean rollbackCommit(String instantTime) {
       while (scannerIterator.hasNext()) {
         Result result = scannerIterator.next();
         currentVersionResults.add(result);
-        statements.add(generateStatement(Bytes.toString(result.getRow()), 0L, rollbackTime - 1));
+        statements.add(generateStatement(fromUTF8Bytes(result.getRow()), 0L, rollbackTime - 1));
 
         if (scannerIterator.hasNext() &&  statements.size() < multiGetBatchSize) {
           continue;

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/index/hbase/TestSparkHoodieHBaseIndex.java

@@ -56,7 +56,6 @@
 import org.apache.hadoop.hbase.client.HTable;
 import org.apache.hadoop.hbase.client.Put;
 import org.apache.hadoop.hbase.client.Result;
-import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.spark.api.java.JavaRDD;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.AfterEach;
@@ -84,6 +83,7 @@
 import static org.apache.hadoop.hbase.HConstants.ZOOKEEPER_CLIENT_PORT;
 import static org.apache.hadoop.hbase.HConstants.ZOOKEEPER_QUORUM;
 import static org.apache.hadoop.hbase.HConstants.ZOOKEEPER_ZNODE_PARENT;
+import static org.apache.hudi.common.util.StringUtils.getUTF8Bytes;
 import static org.apache.hudi.testutils.Assertions.assertNoWriteErrors;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
@@ -124,7 +124,7 @@ public static void init() throws Exception {
     utility = new HBaseTestingUtility(hbaseConfig);
     utility.startMiniCluster();
     hbaseConfig = utility.getConnection().getConfiguration();
-    utility.createTable(TableName.valueOf(TABLE_NAME), Bytes.toBytes("_s"),2);
+    utility.createTable(TableName.valueOf(TABLE_NAME), getUTF8Bytes("_s"), 2);
   }
 
   @AfterAll

hudi-common/src/main/java/org/apache/hudi/avro/GenericAvroSerializer.java

@@ -35,9 +35,9 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
-import java.nio.charset.StandardCharsets;
 import java.util.HashMap;
 
+import static org.apache.hudi.common.util.StringUtils.fromUTF8Bytes;
 import static org.apache.hudi.common.util.StringUtils.getUTF8Bytes;
 
 
@@ -81,7 +81,7 @@ private Schema getSchema(byte[] schemaBytes) {
     if (schemaCache.containsKey(schemaByteBuffer)) {
       return schemaCache.get(schemaByteBuffer);
     } else {
-      String schema = new String(schemaBytes, StandardCharsets.UTF_8);
+      String schema = fromUTF8Bytes(schemaBytes);
       Schema parsedSchema = new Schema.Parser().parse(schema);
       schemaCache.put(schemaByteBuffer, parsedSchema);
       return parsedSchema;

hudi-common/src/main/java/org/apache/hudi/common/bootstrap/index/HFileBootstrapIndex.java

@@ -53,7 +53,6 @@
 import org.apache.hadoop.hbase.io.hfile.HFileContext;
 import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
 import org.apache.hadoop.hbase.io.hfile.HFileScanner;
-import org.apache.hadoop.hbase.util.Bytes;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -100,7 +99,7 @@ public class HFileBootstrapIndex extends BootstrapIndex {
 
   // Additional Metadata written to HFiles.
   public static final String INDEX_INFO_KEY_STRING = "INDEX_INFO";
-  public static final byte[] INDEX_INFO_KEY = Bytes.toBytes(INDEX_INFO_KEY_STRING);
+  public static final byte[] INDEX_INFO_KEY = getUTF8Bytes(INDEX_INFO_KEY_STRING);
 
   private final boolean isPresent;
 
@@ -513,11 +512,11 @@ private <T> List<T> getAllKeys(HFileScanner scanner, Function<String, T> convert
     @Override
     public List<BootstrapFileMapping> getSourceFileMappingForPartition(String partition) {
       try (HFileScanner scanner = partitionIndexReader().getScanner(true, false)) {
-        KeyValue keyValue = new KeyValue(Bytes.toBytes(getPartitionKey(partition)), new byte[0], new byte[0],
+        KeyValue keyValue = new KeyValue(getUTF8Bytes(getPartitionKey(partition)), new byte[0], new byte[0],
             HConstants.LATEST_TIMESTAMP, KeyValue.Type.Put, new byte[0]);
         if (scanner.seekTo(keyValue) == 0) {
           ByteBuffer readValue = scanner.getValue();
-          byte[] valBytes = Bytes.toBytes(readValue);
+          byte[] valBytes = IOUtils.toBytes(readValue);
           HoodieBootstrapPartitionMetadata metadata =
               TimelineMetadataUtils.deserializeAvroMetadata(valBytes, HoodieBootstrapPartitionMetadata.class);
           return metadata.getFileIdToBootstrapFile().entrySet().stream()
@@ -546,11 +545,11 @@ public Map<HoodieFileGroupId, BootstrapFileMapping> getSourceFileMappingForFileI
       Collections.sort(fileGroupIds);
       try (HFileScanner scanner = fileIdIndexReader().getScanner(true, false)) {
         for (HoodieFileGroupId fileGroupId : fileGroupIds) {
-          KeyValue keyValue = new KeyValue(Bytes.toBytes(getFileGroupKey(fileGroupId)), new byte[0], new byte[0],
+          KeyValue keyValue = new KeyValue(getUTF8Bytes(getFileGroupKey(fileGroupId)), new byte[0], new byte[0],
               HConstants.LATEST_TIMESTAMP, KeyValue.Type.Put, new byte[0]);
           if (scanner.seekTo(keyValue) == 0) {
             ByteBuffer readValue = scanner.getValue();
-            byte[] valBytes = Bytes.toBytes(readValue);
+            byte[] valBytes = IOUtils.toBytes(readValue);
             HoodieBootstrapFilePartitionInfo fileInfo = TimelineMetadataUtils.deserializeAvroMetadata(valBytes,
                 HoodieBootstrapFilePartitionInfo.class);
             BootstrapFileMapping mapping = new BootstrapFileMapping(bootstrapBasePath,
@@ -639,7 +638,7 @@ private void writeNextPartition(String partitionPath, String bootstrapPartitionP
         Option<byte[]> bytes = TimelineMetadataUtils.serializeAvroMetadata(bootstrapPartitionMetadata, HoodieBootstrapPartitionMetadata.class);
         if (bytes.isPresent()) {
           indexByPartitionWriter
-              .append(new KeyValue(Bytes.toBytes(getPartitionKey(partitionPath)), new byte[0], new byte[0],
+              .append(new KeyValue(getUTF8Bytes(getPartitionKey(partitionPath)), new byte[0], new byte[0],
                   HConstants.LATEST_TIMESTAMP, KeyValue.Type.Put, bytes.get()));
           numPartitionKeysAdded++;
         }

hudi-common/src/main/java/org/apache/hudi/common/model/HoodieCommitMetadata.java

@@ -36,7 +36,6 @@
 
 import java.io.IOException;
 import java.io.Serializable;
-import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
@@ -48,6 +47,7 @@
 
 import static org.apache.hudi.common.table.timeline.MetadataConversionUtils.convertCommitMetadataToJsonBytes;
 import static org.apache.hudi.common.table.timeline.TimelineMetadataUtils.deserializeCommitMetadata;
+import static org.apache.hudi.common.util.StringUtils.fromUTF8Bytes;
 
 /**
  * All the metadata that gets stored along with a commit.
@@ -249,9 +249,8 @@ public static <T> T fromJsonString(String jsonStr, Class<T> clazz) throws Except
   // TODO: refactor this method to avoid doing the json tree walking (HUDI-4822).
   public static Option<Pair<String, List<String>>> getFileSliceForFileGroupFromDeltaCommit(byte[] bytes, HoodieFileGroupId fileGroupId) {
     try {
-      String jsonStr = new String(
-          convertCommitMetadataToJsonBytes(deserializeCommitMetadata(bytes), org.apache.hudi.avro.model.HoodieCommitMetadata.class),
-          StandardCharsets.UTF_8);
+      String jsonStr = fromUTF8Bytes(
+          convertCommitMetadataToJsonBytes(deserializeCommitMetadata(bytes), org.apache.hudi.avro.model.HoodieCommitMetadata.class));
       if (jsonStr.isEmpty()) {
         return Option.empty();
       }
@@ -517,9 +516,8 @@ public static <T> T fromBytes(byte[] bytes, Class<T> clazz) throws IOException {
         return clazz.newInstance();
       }
       return fromJsonString(
-          new String(
-              convertCommitMetadataToJsonBytes(deserializeCommitMetadata(bytes), org.apache.hudi.avro.model.HoodieCommitMetadata.class),
-              StandardCharsets.UTF_8),
+          fromUTF8Bytes(
+              convertCommitMetadataToJsonBytes(deserializeCommitMetadata(bytes), org.apache.hudi.avro.model.HoodieCommitMetadata.class)),
           clazz);
     } catch (Exception e) {
       throw new IOException("unable to read commit metadata for bytes length: " + bytes.length, e);

hudi-common/src/main/java/org/apache/hudi/common/model/HoodieConsistentHashingMetadata.java

@@ -31,11 +31,11 @@
 import java.io.IOException;
 import java.io.Serializable;
 import java.nio.ByteBuffer;
-import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.UUID;
 
+import static org.apache.hudi.common.util.StringUtils.fromUTF8Bytes;
 import static org.apache.hudi.common.util.StringUtils.getUTF8Bytes;
 
 /**
@@ -159,7 +159,7 @@ public byte[] toBytes() throws IOException {
 
   public static HoodieConsistentHashingMetadata fromBytes(byte[] bytes) throws IOException {
     try {
-      return fromJsonString(new String(bytes, StandardCharsets.UTF_8), HoodieConsistentHashingMetadata.class);
+      return fromJsonString(fromUTF8Bytes(bytes), HoodieConsistentHashingMetadata.class);
     } catch (Exception e) {
       throw new IOException("unable to read hashing metadata", e);
     }

hudi-common/src/main/java/org/apache/hudi/common/model/HoodieReplaceCommitMetadata.java

@@ -25,14 +25,14 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
-import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
 import static org.apache.hudi.common.table.timeline.MetadataConversionUtils.convertCommitMetadataToJsonBytes;
 import static org.apache.hudi.common.table.timeline.TimelineMetadataUtils.deserializeReplaceCommitMetadata;
+import static org.apache.hudi.common.util.StringUtils.fromUTF8Bytes;
 
 /**
  * All the metadata that gets stored along with a commit.
@@ -117,9 +117,8 @@ public static <T> T fromBytes(byte[] bytes, Class<T> clazz) throws IOException {
         return clazz.newInstance();
       }
       return fromJsonString(
-          new String(
-              convertCommitMetadataToJsonBytes(deserializeReplaceCommitMetadata(bytes), org.apache.hudi.avro.model.HoodieReplaceCommitMetadata.class),
-              StandardCharsets.UTF_8),
+          fromUTF8Bytes(
+              convertCommitMetadataToJsonBytes(deserializeReplaceCommitMetadata(bytes), org.apache.hudi.avro.model.HoodieReplaceCommitMetadata.class)),
           clazz);
     } catch (Exception e) {
       throw new IOException("unable to read commit metadata for bytes length: " + bytes.length, e);

hudi-common/src/main/java/org/apache/hudi/common/model/debezium/PostgresDebeziumAvroPayload.java

@@ -30,10 +30,11 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
-import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.Properties;
 
+import static org.apache.hudi.common.util.StringUtils.fromUTF8Bytes;
+
 /**
  * Provides support for seamlessly applying changes captured via Debezium for PostgresDB.
  * <p>
@@ -141,7 +142,7 @@ private boolean containsBytesToastedValues(IndexedRecord incomingRecord, Schema.
         || (field.schema().getType() == Schema.Type.UNION && field.schema().getTypes().stream().anyMatch(s -> s.getType() == Schema.Type.BYTES)))
         // Check length first as an optimization
         && ((ByteBuffer) ((GenericData.Record) incomingRecord).get(field.name())).array().length == DEBEZIUM_TOASTED_VALUE.length()
-        && DEBEZIUM_TOASTED_VALUE.equals(new String(((ByteBuffer) ((GenericData.Record) incomingRecord).get(field.name())).array(), StandardCharsets.UTF_8)));
+        && DEBEZIUM_TOASTED_VALUE.equals(fromUTF8Bytes(((ByteBuffer) ((GenericData.Record) incomingRecord).get(field.name())).array())));
   }
 }
 

hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java

@@ -69,7 +69,6 @@
 
 import java.io.IOException;
 import java.io.Serializable;
-import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -84,6 +83,7 @@
 
 import static org.apache.hudi.common.util.ConfigUtils.containsConfigProperty;
 import static org.apache.hudi.common.util.ConfigUtils.getStringWithAltKeys;
+import static org.apache.hudi.common.util.StringUtils.getUTF8Bytes;
 
 /**
  * <code>HoodieTableMetaClient</code> allows to access meta-data about a hoodie table It returns meta-data about
@@ -212,8 +212,7 @@ public void buildFunctionalIndexDefinition(String indexMetaPath,
       functionalIndexMetadata = Option.of(new HoodieFunctionalIndexMetadata(Collections.singletonMap(indexName, functionalIndexDefinition)));
     }
     try {
-      //fs.mkdirs(new Path(indexMetaPath).getParent());
-      FileIOUtils.createFileInPath(fs, new Path(indexMetaPath), Option.of(functionalIndexMetadata.get().toJson().getBytes(StandardCharsets.UTF_8)));
+      FileIOUtils.createFileInPath(fs, new Path(indexMetaPath), Option.of(getUTF8Bytes(functionalIndexMetadata.get().toJson())));
     } catch (IOException e) {
       throw new HoodieIOException("Could not write functional index metadata at path: " + indexMetaPath, e);
     }
@@ -241,7 +240,7 @@ public void updateFunctionalIndexMetadata(HoodieFunctionalIndexMetadata newFunct
     this.functionalIndexMetadata = Option.of(newFunctionalIndexMetadata);
     try {
       // update the index metadata file as well
-      FileIOUtils.createFileInPath(fs, new Path(indexMetaPath), Option.of(functionalIndexMetadata.get().toJson().getBytes(StandardCharsets.UTF_8)));
+      FileIOUtils.createFileInPath(fs, new Path(indexMetaPath), Option.of(getUTF8Bytes(functionalIndexMetadata.get().toJson())));
     } catch (IOException e) {
       throw new HoodieIOException("Could not write functional index metadata at path: " + indexMetaPath, e);
     }

hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFileReader.java

@@ -40,6 +40,7 @@
 import org.apache.hudi.hadoop.fs.SchemeAwareFSDataInputStream;
 import org.apache.hudi.hadoop.fs.TimedFSDataInputStream;
 import org.apache.hudi.internal.schema.InternalSchema;
+import org.apache.hudi.io.util.IOUtils;
 import org.apache.hudi.storage.StorageSchemes;
 
 import org.apache.avro.Schema;
@@ -49,7 +50,6 @@
 import org.apache.hadoop.fs.FSInputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hbase.util.Bytes;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -332,7 +332,7 @@ private long scanForNextAvailableBlockOffset() throws IOException {
       } catch (EOFException e) {
         eof = true;
       }
-      long pos = Bytes.indexOf(dataBuf, HoodieLogFormat.MAGIC);
+      long pos = IOUtils.indexOf(dataBuf, HoodieLogFormat.MAGIC);
       if (pos >= 0) {
         return currentPos + pos;
       }