You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by to...@apache.org on 2012/05/11 21:27:59 UTC
svn commit: r1337342 - in /hbase/branches/0.90: ./
src/main/java/org/apache/hadoop/hbase/ipc/
src/main/java/org/apache/hadoop/hbase/regionserver/ src/main/ruby/hbase/
src/main/ruby/shell/commands/ src/test/java/org/apache/hadoop/hbase/filter/
Author: todd
Date: Fri May 11 19:27:58 2012
New Revision: 1337342
URL: http://svn.apache.org/viewvc?rev=1337342&view=rev
Log:
HBASE-5973 Add ability for potentially long-running IPC calls to abort if client disconnects
Added:
hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/ipc/CallerDisconnectedException.java
hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/ipc/RpcCallContext.java
Modified:
hbase/branches/0.90/CHANGES.txt
hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/ipc/HBaseServer.java
hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
hbase/branches/0.90/src/main/ruby/hbase/table.rb
hbase/branches/0.90/src/main/ruby/shell/commands/scan.rb
hbase/branches/0.90/src/test/java/org/apache/hadoop/hbase/filter/TestFilter.java
Modified: hbase/branches/0.90/CHANGES.txt
URL: http://svn.apache.org/viewvc/hbase/branches/0.90/CHANGES.txt?rev=1337342&r1=1337341&r2=1337342&view=diff
==============================================================================
--- hbase/branches/0.90/CHANGES.txt (original)
+++ hbase/branches/0.90/CHANGES.txt Fri May 11 19:27:58 2012
@@ -37,6 +37,7 @@ Release 0.90.7 - Unreleased
HBASE-5801 [hbck] Hbck should handle case where some regions have different HTD settings in .regioninfo files (0.90 specific) (Jimmy Xiang)
HBASE-5712 Parallelize load of .regioninfo files in diagnostic/repair portion of hbck
HBASE-3691 Add compressor support for 'snappy', google's compressor
+ HBASE-5973 Add ability for potentially long-running IPC calls to abort if client disconnects (backport by David S. Wang)
NEW FEATURE
HBASE-5128 [uber hbck] Online automated repair of table integrity and region consistency problems
Added: hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/ipc/CallerDisconnectedException.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/ipc/CallerDisconnectedException.java?rev=1337342&view=auto
==============================================================================
--- hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/ipc/CallerDisconnectedException.java (added)
+++ hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/ipc/CallerDisconnectedException.java Fri May 11 19:27:58 2012
@@ -0,0 +1,35 @@
+/**
+ * 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.ipc;
+
+import java.io.IOException;
+
+/**
+ * Exception indicating that the remote host making this IPC lost its
+ * IPC connection. This will never be returned back to a client,
+ * but is only used for logging on the server side, etc.
+ */
+public class CallerDisconnectedException extends IOException {
+ public CallerDisconnectedException(String msg) {
+ super(msg);
+ }
+
+ private static final long serialVersionUID = 1L;
+
+
+}
Modified: hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/ipc/HBaseServer.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/ipc/HBaseServer.java?rev=1337342&r1=1337341&r2=1337342&view=diff
==============================================================================
--- hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/ipc/HBaseServer.java (original)
+++ hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/ipc/HBaseServer.java Fri May 11 19:27:58 2012
@@ -217,7 +217,7 @@ public abstract class HBaseServer {
}
/** A call queued for handling. */
- private static class Call {
+ private static class Call implements RpcCallContext {
protected int id; // the client's call id
protected Writable param; // the parameter passed
protected Connection connection; // connection to client
@@ -241,6 +241,16 @@ public abstract class HBaseServer {
public void setResponse(ByteBuffer response) {
this.response = response;
}
+
+ @Override
+ public void throwExceptionIfCallerDisconnected() throws CallerDisconnectedException {
+ if (!connection.channel.isOpen()) {
+ long afterTime = System.currentTimeMillis() - timestamp;
+ throw new CallerDisconnectedException(
+ "Aborting call " + this + " after " + afterTime + " ms, since " +
+ "caller disconnected");
+ }
+ }
}
/** Listens on the socket. Creates jobs for the handler threads*/
@@ -1400,4 +1410,13 @@ public abstract class HBaseServer {
int nBytes = initialRemaining - buf.remaining();
return (nBytes > 0) ? nBytes : ret;
}
+
+ /**
+ * Needed for delayed calls. We need to be able to store the current call
+ * so that we can complete it later.
+ * @return Call the server is currently handling.
+ */
+ public static RpcCallContext getCurrentCall() {
+ return CurCall.get();
+ }
}
Added: hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/ipc/RpcCallContext.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/ipc/RpcCallContext.java?rev=1337342&view=auto
==============================================================================
--- hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/ipc/RpcCallContext.java (added)
+++ hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/ipc/RpcCallContext.java Fri May 11 19:27:58 2012
@@ -0,0 +1,29 @@
+/**
+ * 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.ipc;
+
+public interface RpcCallContext {
+
+ /**
+ * Throw an exception if the caller who made this IPC call has disconnected.
+ * If called from outside the context of IPC, this does nothing.
+ * @throws CallerDisconnectedException
+ */
+ void throwExceptionIfCallerDisconnected() throws CallerDisconnectedException;
+
+}
Modified: hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java?rev=1337342&r1=1337341&r2=1337342&view=diff
==============================================================================
--- hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java (original)
+++ hbase/branches/0.90/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java Fri May 11 19:27:58 2012
@@ -77,6 +77,8 @@ import org.apache.hadoop.hbase.filter.In
import org.apache.hadoop.hbase.io.HeapSize;
import org.apache.hadoop.hbase.io.TimeRange;
import org.apache.hadoop.hbase.io.hfile.BlockCache;
+import org.apache.hadoop.hbase.ipc.HBaseServer;
+import org.apache.hadoop.hbase.ipc.RpcCallContext;
import org.apache.hadoop.hbase.regionserver.wal.HLog;
import org.apache.hadoop.hbase.regionserver.wal.HLogKey;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
@@ -2449,7 +2451,16 @@ public class HRegion implements HeapSize
}
private boolean nextInternal(int limit) throws IOException {
+ RpcCallContext rpcCall = HBaseServer.getCurrentCall();
while (true) {
+ if (rpcCall != null) {
+ // If a user specifies a too-restrictive or too-slow scanner, the
+ // client might time out and disconnect while the server side
+ // is still processing the request. We should abort aggressively
+ // in that case.
+ rpcCall.throwExceptionIfCallerDisconnected();
+ }
+
byte [] currentRow = peekRow();
if (isStopRow(currentRow)) {
if (filter != null && filter.hasFilterRow()) {
Modified: hbase/branches/0.90/src/main/ruby/hbase/table.rb
URL: http://svn.apache.org/viewvc/hbase/branches/0.90/src/main/ruby/hbase/table.rb?rev=1337342&r1=1337341&r2=1337342&view=diff
==============================================================================
--- hbase/branches/0.90/src/main/ruby/hbase/table.rb (original)
+++ hbase/branches/0.90/src/main/ruby/hbase/table.rb Fri May 11 19:27:58 2012
@@ -219,7 +219,8 @@ module Hbase
stoprow = args["STOPROW"]
timestamp = args["TIMESTAMP"]
columns = args["COLUMNS"] || args["COLUMN"] || get_all_columns
- cache = args["CACHE_BLOCKS"] || true
+ cache_blocks = args["CACHE_BLOCKS"] || true
+ cache = args["CACHE"] || 0
versions = args["VERSIONS"] || 1
timerange = args[TIMERANGE]
@@ -238,7 +239,8 @@ module Hbase
columns.each { |c| scan.addColumns(c) }
scan.setFilter(filter) if filter
scan.setTimeStamp(timestamp) if timestamp
- scan.setCacheBlocks(cache)
+ scan.setCacheBlocks(cache_blocks)
+ scan.setCaching(cache) if cache > 0
scan.setMaxVersions(versions) if versions > 1
scan.setTimeRange(timerange[0], timerange[1]) if timerange
else
Modified: hbase/branches/0.90/src/main/ruby/shell/commands/scan.rb
URL: http://svn.apache.org/viewvc/hbase/branches/0.90/src/main/ruby/shell/commands/scan.rb?rev=1337342&r1=1337341&r2=1337342&view=diff
==============================================================================
--- hbase/branches/0.90/src/main/ruby/shell/commands/scan.rb (original)
+++ hbase/branches/0.90/src/main/ruby/shell/commands/scan.rb Fri May 11 19:27:58 2012
@@ -26,7 +26,7 @@ module Shell
Scan a table; pass table name and optionally a dictionary of scanner
specifications. Scanner specifications may include one or more of:
TIMERANGE, FILTER, LIMIT, STARTROW, STOPROW, TIMESTAMP, MAXLENGTH,
-or COLUMNS. If no columns are specified, all columns will be scanned.
+or COLUMNS, CACHE. If no columns are specified, all columns will be scanned.
To scan all members of a column family, leave the qualifier empty as in
'col_family:'.
Modified: hbase/branches/0.90/src/test/java/org/apache/hadoop/hbase/filter/TestFilter.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.90/src/test/java/org/apache/hadoop/hbase/filter/TestFilter.java?rev=1337342&r1=1337341&r2=1337342&view=diff
==============================================================================
--- hbase/branches/0.90/src/test/java/org/apache/hadoop/hbase/filter/TestFilter.java (original)
+++ hbase/branches/0.90/src/test/java/org/apache/hadoop/hbase/filter/TestFilter.java Fri May 11 19:27:58 2012
@@ -20,6 +20,8 @@
package org.apache.hadoop.hbase.filter;
+import java.io.DataInput;
+import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
@@ -44,11 +46,13 @@ import org.apache.hadoop.hbase.regionser
import org.apache.hadoop.hbase.regionserver.InternalScanner;
import org.apache.hadoop.hbase.util.Bytes;
+import com.google.common.base.Throwables;
+
/**
* Test filters at the HRegion doorstep.
*/
public class TestFilter extends HBaseTestCase {
- private final Log LOG = LogFactory.getLog(this.getClass());
+ private final static Log LOG = LogFactory.getLog(TestFilter.class);
private HRegion region;
//
@@ -1511,4 +1515,38 @@ public class TestFilter extends HBaseTes
verifyScanFullNoValues(s, expectedKVs, useLen);
}
}
+
+ /**
+ * Filter which makes sleeps for a second between each row of a scan.
+ * This can be useful for manual testing of bugs like HBASE-5973. For example:
+ * <code>
+ * create 't1', 'f1'
+ * 1.upto(100) { |x| put 't1', 'r' + x.to_s, 'f1:q1', 'hi' }
+ * import org.apache.hadoop.hbase.filter.TestFilter
+ * scan 't1', { FILTER => TestFilter::SlowScanFilter.new(), CACHE => 50 }
+ * </code>
+ */
+ public static class SlowScanFilter extends FilterBase {
+ private static Thread ipcHandlerThread = null;
+
+ @Override
+ public void readFields(DataInput arg0) throws IOException {
+ }
+
+ @Override
+ public void write(DataOutput arg0) throws IOException {
+ }
+
+ @Override
+ public boolean filterRow() {
+ ipcHandlerThread = Thread.currentThread();
+ try {
+ LOG.info("Handler thread " + ipcHandlerThread + " sleeping in filter...");
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ Throwables.propagate(e);
+ }
+ return super.filterRow();
+ }
+ }
}