HBASE-28175: Deep copy RpcLogDetails' param field

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

@@ -82,7 +82,7 @@ public boolean equals(Object o) {
     }
     SlowLogParams that = (SlowLogParams) o;
     return new EqualsBuilder().append(regionName, that.regionName).append(params, that.params)
-      .append("scan", scan).isEquals();
+      .append(scan, that.scan).isEquals();
   }
 
   @Override

hbase-server/src/main/java/org/apache/hadoop/hbase/namequeues/RpcLogDetails.java

@@ -21,7 +21,10 @@
 import org.apache.commons.lang3.builder.ToStringBuilder;
 import org.apache.hadoop.hbase.ipc.RpcCall;
 import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
+import org.apache.hbase.thirdparty.com.google.protobuf.InvalidProtocolBufferException;
 import org.apache.hbase.thirdparty.com.google.protobuf.Message;
 
 /**
@@ -32,8 +35,10 @@ public class RpcLogDetails extends NamedQueuePayload {
 
   public static final int SLOW_LOG_EVENT = 0;
 
+  private static final Logger LOG = LoggerFactory.getLogger(RpcLogDetails.class.getName());
+
   private final RpcCall rpcCall;
-  private final Message param;
+  private Message param;
   private final String clientAddress;
   private final long responseSize;
   private final long blockBytesScanned;
@@ -47,7 +52,6 @@ public RpcLogDetails(RpcCall rpcCall, Message param, String clientAddress, long
     long blockBytesScanned, String className, boolean isSlowLog, boolean isLargeLog) {
     super(SLOW_LOG_EVENT);
     this.rpcCall = rpcCall;
-    this.param = param;
     this.clientAddress = clientAddress;
     this.responseSize = responseSize;
     this.blockBytesScanned = blockBytesScanned;
@@ -60,6 +64,18 @@ public RpcLogDetails(RpcCall rpcCall, Message param, String clientAddress, long
     // would result in corrupted attributes
     this.connectionAttributes = rpcCall.getConnectionAttributes();
     this.requestAttributes = rpcCall.getRequestAttributes();
+
+    // We also need to deep copy the message because the CodedInputStream may be
+    // overwritten before this slow log is consumed. Such overwriting could
+    // cause the slow log payload to be corrupt
+    try {
+      this.param = param.newBuilderForType().mergeFrom(param.toByteArray()).build();
+    } catch (InvalidProtocolBufferException e) {
+      LOG.error("Failed to parse protobuf for message {}", param, e);
+      if (this.param == null) {
+        this.param = param;
+      }
+    }
   }
 
   public RpcCall getRpcCall() {

hbase-server/src/test/java/org/apache/hadoop/hbase/namequeues/TestRpcLogDetails.java

@@ -0,0 +1,265 @@
+/*
+ * 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.namequeues;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Optional;
+import org.apache.hadoop.hbase.CellScanner;
+import org.apache.hadoop.hbase.HBaseClassTestRule;
+import org.apache.hadoop.hbase.ipc.RpcCall;
+import org.apache.hadoop.hbase.ipc.RpcCallback;
+import org.apache.hadoop.hbase.security.User;
+import org.apache.hadoop.hbase.testclassification.RegionServerTests;
+import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.hbase.thirdparty.com.google.protobuf.BlockingService;
+import org.apache.hbase.thirdparty.com.google.protobuf.ByteString;
+import org.apache.hbase.thirdparty.com.google.protobuf.CodedInputStream;
+import org.apache.hbase.thirdparty.com.google.protobuf.Descriptors;
+import org.apache.hbase.thirdparty.com.google.protobuf.Message;
+import org.apache.hbase.thirdparty.com.google.protobuf.UnsafeByteOperations;
+
+import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.RPCProtos;
+
+@Category({ RegionServerTests.class, SmallTests.class })
+public class TestRpcLogDetails {
+
+  @ClassRule
+  public static final HBaseClassTestRule CLASS_RULE =
+    HBaseClassTestRule.forClass(TestRpcLogDetails.class);
+
+  private final ClientProtos.Scan scan =
+    ClientProtos.Scan.newBuilder().setStartRow(ByteString.copyFrom(Bytes.toBytes("abc")))
+      .setStopRow(ByteString.copyFrom(Bytes.toBytes("xyz"))).build();
+  private final ClientProtos.Scan otherScan =
+    ClientProtos.Scan.newBuilder().setStartRow(ByteString.copyFrom(Bytes.toBytes("def")))
+      .setStopRow(ByteString.copyFrom(Bytes.toBytes("uvw"))).build();
+  private final ClientProtos.ScanRequest scanRequest = ClientProtos.ScanRequest
+    .newBuilder(ClientProtos.ScanRequest.getDefaultInstance()).setScan(scan).build();
+  private final ClientProtos.ScanRequest otherScanRequest = ClientProtos.ScanRequest
+    .newBuilder(ClientProtos.ScanRequest.getDefaultInstance()).setScan(otherScan).build();
+
+  @Test
+  public void itDeepCopiesRpcLogDetailsParams() throws IOException {
+    ByteBuffer buffer = ByteBuffer.allocate(scanRequest.toByteArray().length);
+    CodedInputStream cis = UnsafeByteOperations.unsafeWrap(buffer).newCodedInput();
+    cis.enableAliasing(true);
+    buffer.put(scanRequest.toByteArray());
+    Message.Builder messageBuilder = ClientProtos.ScanRequest.newBuilder();
+    ProtobufUtil.mergeFrom(messageBuilder, cis, buffer.capacity());
+    Message message = messageBuilder.build();
+    RpcLogDetails rpcLogDetails =
+      new RpcLogDetails(getRpcCall(message), message, null, 0L, 0L, null, true, false);
+
+    // log's scan should be equal
+    ClientProtos.Scan logScan = ((ClientProtos.ScanRequest) rpcLogDetails.getParam()).getScan();
+    assertEquals(logScan, scan);
+
+    // ensure we have a different byte array for testing
+    assertFalse(Arrays.equals(scanRequest.toByteArray(), otherScanRequest.toByteArray()));
+
+    // corrupt the underlying buffer
+    buffer.position(0);
+    buffer.put(otherScanRequest.toByteArray(), 0, otherScanRequest.toByteArray().length);
+    assertArrayEquals(otherScanRequest.toByteArray(), buffer.array());
+
+    // log scan should still be original scan
+    assertEquals(logScan, scan);
+  }
+
+  @SuppressWarnings("checkstyle:methodlength")
+  private static RpcCall getRpcCall(Message message) {
+    RpcCall rpcCall = new RpcCall() {
+      @Override
+      public BlockingService getService() {
+        return null;
+      }
+
+      @Override
+      public Descriptors.MethodDescriptor getMethod() {
+        return null;
+      }
+
+      @Override
+      public Message getParam() {
+        return message;
+      }
+
+      @Override
+      public CellScanner getCellScanner() {
+        return null;
+      }
+
+      @Override
+      public long getReceiveTime() {
+        return 0;
+      }
+
+      @Override
+      public long getStartTime() {
+        return 0;
+      }
+
+      @Override
+      public void setStartTime(long startTime) {
+      }
+
+      @Override
+      public int getTimeout() {
+        return 0;
+      }
+
+      @Override
+      public int getPriority() {
+        return 0;
+      }
+
+      @Override
+      public long getDeadline() {
+        return 0;
+      }
+
+      @Override
+      public long getSize() {
+        return 0;
+      }
+
+      @Override
+      public RPCProtos.RequestHeader getHeader() {
+        return null;
+      }
+
+      @Override
+      public Map<String, byte[]> getConnectionAttributes() {
+        return Collections.emptyMap();
+      }
+
+      @Override
+      public Map<String, byte[]> getRequestAttributes() {
+        return Collections.emptyMap();
+      }
+
+      @Override
+      public byte[] getRequestAttribute(String key) {
+        return null;
+      }
+
+      @Override
+      public int getRemotePort() {
+        return 0;
+      }
+
+      @Override
+      public void setResponse(Message param, CellScanner cells, Throwable errorThrowable,
+        String error) {
+      }
+
+      @Override
+      public void sendResponseIfReady() throws IOException {
+      }
+
+      @Override
+      public void cleanup() {
+      }
+
+      @Override
+      public String toShortString() {
+        return null;
+      }
+
+      @Override
+      public long disconnectSince() {
+        return 0;
+      }
+
+      @Override
+      public boolean isClientCellBlockSupported() {
+        return false;
+      }
+
+      @Override
+      public Optional<User> getRequestUser() {
+        return null;
+      }
+
+      @Override
+      public InetAddress getRemoteAddress() {
+        return null;
+      }
+
+      @Override
+      public HBaseProtos.VersionInfo getClientVersionInfo() {
+        return null;
+      }
+
+      @Override
+      public void setCallBack(RpcCallback callback) {
+      }
+
+      @Override
+      public boolean isRetryImmediatelySupported() {
+        return false;
+      }
+
+      @Override
+      public long getResponseCellSize() {
+        return 0;
+      }
+
+      @Override
+      public void incrementResponseCellSize(long cellSize) {
+      }
+
+      @Override
+      public long getBlockBytesScanned() {
+        return 0;
+      }
+
+      @Override
+      public void incrementBlockBytesScanned(long blockSize) {
+      }
+
+      @Override
+      public long getResponseExceptionSize() {
+        return 0;
+      }
+
+      @Override
+      public void incrementResponseExceptionSize(long exceptionSize) {
+      }
+    };
+    return rpcCall;
+  }
+
+}