Revert "HBASE-28001: Add request attribute support to BufferedMutator (apache#6076)"

This reverts commit 7d019d9.

Author: ndimiduk

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncBufferedMutator.java

@@ -20,7 +20,6 @@
 import java.io.Closeable;
 import java.util.Collections;
 import java.util.List;
-import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import org.apache.hadoop.conf.Configuration;
@@ -94,11 +93,4 @@ default CompletableFuture<Void> mutate(Mutation mutation) {
   default long getPeriodicalFlushTimeout(TimeUnit unit) {
     throw new UnsupportedOperationException("Not implemented");
   }
-
-  /**
-   * Returns the rpc request attributes.
-   */
-  default Map<String, byte[]> getRequestAttributes() {
-    return Collections.emptyMap();
-  }
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncBufferedMutatorBuilder.java

@@ -19,7 +19,6 @@
 
 import static org.apache.hadoop.hbase.client.ConnectionUtils.retries2Attempts;
 
-import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import org.apache.yetus.audience.InterfaceAudience;
 
@@ -104,16 +103,6 @@ default AsyncBufferedMutatorBuilder setMaxRetries(int maxRetries) {
    */
   AsyncBufferedMutatorBuilder setMaxKeyValueSize(int maxKeyValueSize);
 
-  /**
-   * Set a rpc request attribute.
-   */
-  AsyncBufferedMutatorBuilder setRequestAttribute(String key, byte[] value);
-
-  /**
-   * Set multiple rpc request attributes.
-   */
-  AsyncBufferedMutatorBuilder setRequestAttributes(Map<String, byte[]> requestAttributes);
-
   /**
    * Create the {@link AsyncBufferedMutator} instance.
    */

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncBufferedMutatorBuilderImpl.java

@@ -17,7 +17,6 @@
  */
 package org.apache.hadoop.hbase.client;
 
-import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import org.apache.yetus.audience.InterfaceAudience;
 
@@ -79,20 +78,6 @@ public AsyncBufferedMutatorBuilder setStartLogErrorsCnt(int startLogErrorsCnt) {
     return this;
   }
 
-  @Override
-  public AsyncBufferedMutatorBuilder setRequestAttribute(String key, byte[] value) {
-    tableBuilder.setRequestAttribute(key, value);
-    return this;
-  }
-
-  @Override
-  public AsyncBufferedMutatorBuilder setRequestAttributes(Map<String, byte[]> requestAttributes) {
-    for (Map.Entry<String, byte[]> requestAttribute : requestAttributes.entrySet()) {
-      tableBuilder.setRequestAttribute(requestAttribute.getKey(), requestAttribute.getValue());
-    }
-    return this;
-  }
-
   @Override
   public AsyncBufferedMutatorBuilder setWriteBufferSize(long writeBufferSize) {
     Preconditions.checkArgument(writeBufferSize > 0, "writeBufferSize %d must be > 0",

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncBufferedMutatorImpl.java

@@ -24,7 +24,6 @@
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
-import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
@@ -171,9 +170,4 @@ public long getWriteBufferSize() {
   public long getPeriodicalFlushTimeout(TimeUnit unit) {
     return unit.convert(periodicFlushTimeoutNs, TimeUnit.NANOSECONDS);
   }
-
-  @Override
-  public Map<String, byte[]> getRequestAttributes() {
-    return table.getRequestAttributes();
-  }
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncTable.java

@@ -22,12 +22,12 @@
 import static org.apache.hadoop.hbase.util.FutureUtils.allOf;
 
 import com.google.protobuf.RpcChannel;
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
+import org.apache.commons.lang3.NotImplementedException;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.CompareOperator;
 import org.apache.hadoop.hbase.TableName;
@@ -117,7 +117,7 @@ public interface AsyncTable<C extends ScanResultConsumerBase> {
    * @return a map of request attributes supplied by the client
    */
   default Map<String, byte[]> getRequestAttributes() {
-    return Collections.emptyMap();
+    throw new NotImplementedException("Add an implementation!");
   }
 
   /**

hbase-client/src/main/java/org/apache/hadoop/hbase/client/BufferedMutator.java

@@ -19,9 +19,7 @@
 
 import java.io.Closeable;
 import java.io.IOException;
-import java.util.Collections;
 import java.util.List;
-import java.util.Map;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.yetus.audience.InterfaceAudience;
@@ -196,13 +194,6 @@ default void setOperationTimeout(int timeout) {
       "The BufferedMutator::setOperationTimeout has not been implemented");
   }
 
-  /**
-   * Returns the rpc request attributes.
-   */
-  default Map<String, byte[]> getRequestAttributes() {
-    return Collections.emptyMap();
-  }
-
   /**
    * Listens for asynchronous exceptions on a {@link BufferedMutator}.
    */

hbase-client/src/main/java/org/apache/hadoop/hbase/client/BufferedMutatorImpl.java

@@ -26,7 +26,6 @@
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
-import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Timer;
 import java.util.TimerTask;
@@ -90,7 +89,6 @@ public class BufferedMutatorImpl implements BufferedMutator {
   private final ExecutorService pool;
   private final AtomicInteger rpcTimeout;
   private final AtomicInteger operationTimeout;
-  private final Map<String, byte[]> requestAttributes;
   private final boolean cleanupPoolOnClose;
   private volatile boolean closed = false;
   private final AsyncProcess ap;
@@ -137,9 +135,6 @@ public class BufferedMutatorImpl implements BufferedMutator {
     this.operationTimeout = new AtomicInteger(params.getOperationTimeout() != UNSET
       ? params.getOperationTimeout()
       : conn.getConnectionConfiguration().getOperationTimeout());
-
-    this.requestAttributes = params.getRequestAttributes();
-
     this.ap = ap;
   }
 
@@ -257,8 +252,7 @@ public synchronized void close() throws IOException {
 
   private AsyncProcessTask createTask(QueueRowAccess access) {
     return new AsyncProcessTask(AsyncProcessTask.newBuilder().setPool(pool).setTableName(tableName)
-      .setRowAccess(access).setSubmittedRows(AsyncProcessTask.SubmittedRows.AT_LEAST_ONE)
-      .setRequestAttributes(requestAttributes).build()) {
+      .setRowAccess(access).setSubmittedRows(AsyncProcessTask.SubmittedRows.AT_LEAST_ONE).build()) {
       @Override
       public int getRpcTimeout() {
         return rpcTimeout.get();
@@ -397,11 +391,6 @@ public void setOperationTimeout(int operationTimeout) {
     this.operationTimeout.set(operationTimeout);
   }
 
-  @Override
-  public Map<String, byte[]> getRequestAttributes() {
-    return requestAttributes;
-  }
-
   long getCurrentWriteBufferSize() {
     return currentWriteBufferSize.get();
   }

hbase-client/src/main/java/org/apache/hadoop/hbase/client/BufferedMutatorParams.java

@@ -17,9 +17,6 @@
  */
 package org.apache.hadoop.hbase.client;
 
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.yetus.audience.InterfaceAudience;
@@ -41,7 +38,6 @@ public class BufferedMutatorParams implements Cloneable {
   private String implementationClassName = null;
   private int rpcTimeout = UNSET;
   private int operationTimeout = UNSET;
-  protected Map<String, byte[]> requestAttributes = Collections.emptyMap();
   private BufferedMutator.ExceptionListener listener = new BufferedMutator.ExceptionListener() {
     @Override
     public void onException(RetriesExhaustedWithDetailsException exception,
@@ -89,18 +85,6 @@ public int getOperationTimeout() {
     return operationTimeout;
   }
 
-  public BufferedMutatorParams setRequestAttribute(String key, byte[] value) {
-    if (requestAttributes.isEmpty()) {
-      requestAttributes = new HashMap<>();
-    }
-    requestAttributes.put(key, value);
-    return this;
-  }
-
-  public Map<String, byte[]> getRequestAttributes() {
-    return requestAttributes;
-  }
-
   /**
    * Override the write buffer size specified by the provided {@link Connection}'s
    * {@link org.apache.hadoop.conf.Configuration} instance, via the configuration key

hbase-examples/src/main/java/org/apache/hadoop/hbase/client/example/BufferedMutatorExample.java

@@ -19,19 +19,15 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
-import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.conf.Configured;
-import org.apache.hadoop.hbase.AuthUtil;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.BufferedMutator;
 import org.apache.hadoop.hbase.client.BufferedMutatorParams;
@@ -71,18 +67,12 @@ public void onException(RetriesExhaustedWithDetailsException e, BufferedMutator
         }
       }
     };
-    BufferedMutatorParams params = new BufferedMutatorParams(TABLE).listener(listener)
-      .setRequestAttribute("requestInfo", Bytes.toBytes("bar"));
+    BufferedMutatorParams params = new BufferedMutatorParams(TABLE).listener(listener);
 
     //
     // step 1: create a single Connection and a BufferedMutator, shared by all worker threads.
     //
-    Map<String, byte[]> connectionAttributes = new HashMap<>();
-    connectionAttributes.put("clientId", Bytes.toBytes("foo"));
-    Configuration conf = getConf();
-    try (
-      final Connection conn = ConnectionFactory.createConnection(conf, null,
-        AuthUtil.loginClient(conf), connectionAttributes);
+    try (final Connection conn = ConnectionFactory.createConnection(getConf());
       final BufferedMutator mutator = conn.getBufferedMutator(params)) {
 
       /** worker pool that operates on BufferedTable instances */