HBASE-26087 JVM crash when displaying RPC params by MonitoredRPCHandler

Author: sunhelly

hbase-server/src/main/java/org/apache/hadoop/hbase/monitoring/MonitoredRPCHandlerImpl.java

@@ -53,11 +53,20 @@ public MonitoredRPCHandlerImpl() {
 
   @Override
   public synchronized MonitoredRPCHandlerImpl clone() {
-    return (MonitoredRPCHandlerImpl) super.clone();
+    MonitoredRPCHandlerImpl clone = (MonitoredRPCHandlerImpl) super.clone();
+    return wrap(clone);
+  }
+
+  public synchronized WrappedMonitoredRPCHandlerImpl wrap(MonitoredRPCHandlerImpl
+    monitoredRPCHandler) {
+    WrappedMonitoredRPCHandlerImpl wrappedMonitoredRPCHandler =
+      new WrappedMonitoredRPCHandlerImpl(monitoredRPCHandler);
+    wrappedMonitoredRPCHandler.setCallInfoMap(toMap());
+    return wrappedMonitoredRPCHandler;
   }
 
   /**
-   * Gets the status of this handler; if it is currently servicing an RPC, 
+   * Gets the status of this handler; if it is currently servicing an RPC,
    * this status will include the RPC information.
    * @return a String describing the current status.
    */

hbase-server/src/main/java/org/apache/hadoop/hbase/monitoring/WrappedMonitoredRPCHandlerImpl.java

@@ -0,0 +1,126 @@
+/*
+ * 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.monitoring;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.yetus.audience.InterfaceAudience;
+
+/**
+ * A wrap of MonitoredRPCHandlerImpl, it has the call info from the call param.
+ * Please use this class to avoid read offheap param data.
+ */
+@InterfaceAudience.Private
+public class WrappedMonitoredRPCHandlerImpl extends MonitoredRPCHandlerImpl
+  implements MonitoredTask {
+  private Map<String, Object> callInfoMap = new HashMap<>();
+  private MonitoredRPCHandlerImpl monitoredRPCHandler;
+
+  public WrappedMonitoredRPCHandlerImpl(MonitoredRPCHandlerImpl monitoredRPCHandler) {
+    this.monitoredRPCHandler = monitoredRPCHandler;
+  }
+
+  public void setCallInfoMap(Map<String, Object> callInfoMap) {
+    this.callInfoMap = callInfoMap;
+  }
+
+  @Override
+  public long getStartTime() {
+    return monitoredRPCHandler.getStartTime();
+  }
+
+  @Override
+  public String getDescription() {
+    return monitoredRPCHandler.getDescription();
+  }
+
+  @Override
+  public String getStatus() {
+    return monitoredRPCHandler.getStatus();
+  }
+
+  @Override
+  public long getStatusTime() {
+    return monitoredRPCHandler.getStatusTime();
+  }
+
+  @Override
+  public State getState() {
+    return monitoredRPCHandler.getState();
+  }
+
+  @Override
+  public long getStateTime() {
+    return monitoredRPCHandler.getStateTime();
+  }
+
+  @Override
+  public long getCompletionTimestamp() {
+    return monitoredRPCHandler.getCompletionTimestamp();
+  }
+
+  @Override
+  public String getRPC() {
+    return monitoredRPCHandler.getRPC(false);
+  }
+
+  @Override
+  public String getRPC(boolean withParams) {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public long getRPCStartTime() {
+    return monitoredRPCHandler.getRPCStartTime();
+  }
+
+  @Override
+  public long getRPCQueueTime() {
+    return monitoredRPCHandler.getRPCQueueTime();
+  }
+
+  @Override
+  public boolean isOperationRunning() {
+    return monitoredRPCHandler.isOperationRunning();
+  }
+
+  @Override
+  public long getRPCPacketLength() {
+    return monitoredRPCHandler.getRPCPacketLength();
+  }
+
+  @Override
+  public String getClient() {
+    return monitoredRPCHandler.getClient();
+  }
+
+  @Override
+  public boolean isRPCRunning() {
+    return monitoredRPCHandler.isRPCRunning();
+  }
+
+  @Override
+  public Map<String, Object> toMap() {
+    return callInfoMap;
+  }
+
+  @Override
+  public String toString() {
+    return monitoredRPCHandler.toString();
+  }
+}

hbase-server/src/test/java/org/apache/hadoop/hbase/monitoring/TestTaskMonitor.java

@@ -17,10 +17,13 @@
  */
 package org.apache.hadoop.hbase.monitoring;
 
-import static org.junit.Assert.*;
-
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HBaseClassTestRule;
@@ -34,9 +37,12 @@
 import org.junit.ClassRule;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 @Category({MiscTests.class, SmallTests.class})
 public class TestTaskMonitor {
+  private static final Logger LOG = LoggerFactory.getLogger(TestTaskMonitor.class);
 
   @ClassRule
   public static final HBaseClassTestRule CLASS_RULE =
@@ -226,5 +232,66 @@ public void testStatusJournal() {
     assertEquals("status3", task.getStatusJournal().get(1).getStatus());
     tm.shutdown();
   }
+
+  @Test
+  public void testClone() throws Exception {
+    MonitoredRPCHandlerImpl monitor = new MonitoredRPCHandlerImpl();
+    monitor.abort("abort RPC");
+    TestParam testParam = new TestParam("param1");
+    monitor.setRPC("method1", new Object[]{ testParam }, 0);
+    MonitoredRPCHandlerImpl clone = monitor.clone();
+    assertEquals(clone.getDescription(), monitor.getDescription());
+    assertEquals(clone.getState(), monitor.getState());
+    assertEquals(clone.getStatus(), monitor.getStatus());
+    assertEquals(clone.toString(), monitor.toString());
+    assertEquals(clone.toMap(), monitor.toMap());
+    assertEquals(clone.toJSON(), monitor.toJSON());
+
+    // mark complete and make param dirty
+    monitor.markComplete("complete RPC");
+    testParam.setParam("dirtyParam");
+    assertEquals(clone.getDescription(), monitor.getDescription());
+    assertNotEquals(clone.getState(), monitor.getState());
+    assertNotEquals(clone.getStatus(), monitor.getStatus());
+    monitor.setState(MonitoredTask.State.RUNNING);
+    try {
+      // when markComplete, the param in monitor is set null, so toMap should fail here
+      monitor.toMap();
+      fail("Should not call toMap successfully, because param=null");
+    } catch (Exception e) {
+    }
+    // the param of clone monitor should not be dirty
+    assertNotEquals("[dirtyString]",
+      String.valueOf(((Map<String, Object>) clone.toMap().get("rpcCall")).get("params")));
+
+    monitor.resume("resume");
+    monitor.setRPC("method2", new Object[]{new TestParam("param2")}, 1);
+    assertNotEquals(((Map<String, Object>) clone.toMap().get("rpcCall")).get("params"),
+      ((Map<String, Object>) monitor.toMap().get("rpcCall")).get(
+        "params"));
+    LOG.info(String.valueOf(clone.toMap()));
+    LOG.info(String.valueOf(monitor.toMap()));
+    assertNotEquals(clone.toString(), monitor.toString());
+    assertNotEquals(clone.getRPCQueueTime(), monitor.getRPCQueueTime());
+    assertNotEquals(clone.toMap(), monitor.toMap());
+    assertNotEquals(clone.toJSON(), monitor.toJSON());
+  }
+
+  private class TestParam {
+    public String param = null;
+
+    public TestParam(String param) {
+      this.param = param;
+    }
+
+    public void setParam(String param) {
+      this.param = param;
+    }
+
+    @Override
+    public String toString() {
+      return param;
+    }
+  }
 }