Apache
/
hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HBASE-19471 Fixed remaining Checkstyle errors in hbase-thrift

This commit is contained in:
Jan Hentschel 2017-12-09 20:17:23 +01:00
parent 228d7a5a46
commit 830179600d
17 changed files with 750 additions and 678 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
hbase-checkstyle/src/main/resources/hbase/checkstyle-suppressions.xml
View File

@ -36,4 +36,5 @@
<suppress checks="MagicNumberCheck" files=".*/src/test/.*\.java"/>
<suppress checks="VisibilityModifier" files=".*/src/test/.*\.java"/>
<suppress checks="InterfaceIsTypeCheck" files=".*/src/main/.*\.java"/>
<suppress checks="EmptyBlockCheck" files="TBoundedThreadPoolServer.java"/>
</suppressions>

16
hbase-thrift/pom.xml
View File

@ -139,6 +139,22 @@
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-source-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-checkstyle-plugin</artifactId>
<executions>
<execution>
<id>checkstyle</id>
<phase>validate</phase>
<goals>
<goal>check</goal>
</goals>
<configuration>
<failOnViolation>true</failOnViolation>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
<pluginManagement>
<plugins>

2
hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift/HbaseHandlerMetricsProxy.java
View File

@ -34,7 +34,7 @@ import org.slf4j.LoggerFactory;
* time of each call to ThriftMetrics.
*/
@InterfaceAudience.Private
public class HbaseHandlerMetricsProxy implements InvocationHandler {
public final class HbaseHandlerMetricsProxy implements InvocationHandler {
private static final Logger LOG = LoggerFactory.getLogger(
HbaseHandlerMetricsProxy.class);

56
hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift/IncrementCoalescer.java
View File

@ -117,14 +117,30 @@ public class IncrementCoalescer implements IncrementCoalescerMBean {
@Override
public boolean equals(Object obj) {
if (this == obj) return true;
if (obj == null) return false;
if (getClass() != obj.getClass()) return false;
if (this == obj) {
return true;
}
if (obj == null) {
return false;
}
if (getClass() != obj.getClass()) {
return false;
}
FullyQualifiedRow other = (FullyQualifiedRow) obj;
if (!Arrays.equals(family, other.family)) return false;
if (!Arrays.equals(qualifier, other.qualifier)) return false;
if (!Arrays.equals(rowKey, other.rowKey)) return false;
if (!Arrays.equals(table, other.table)) return false;
if (!Arrays.equals(family, other.family)) {
return false;
}
if (!Arrays.equals(qualifier, other.qualifier)) {
return false;
}
if (!Arrays.equals(rowKey, other.rowKey)) {
return false;
}
if (!Arrays.equals(table, other.table)) {
return false;
}
return true;
}
@ -144,8 +160,14 @@ public class IncrementCoalescer implements IncrementCoalescerMBean {
public Thread newThread(Runnable r) {
Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
if (!t.isDaemon()) t.setDaemon(true);
if (t.getPriority() != Thread.NORM_PRIORITY) t.setPriority(Thread.NORM_PRIORITY);
if (!t.isDaemon()) {
t.setDaemon(true);
}
if (t.getPriority() != Thread.NORM_PRIORITY) {
t.setPriority(Thread.NORM_PRIORITY);
}
return t;
}
}
@ -191,13 +213,16 @@ public class IncrementCoalescer implements IncrementCoalescerMBean {
for (TIncrement tinc : incs) {
internalQueueTincrement(tinc);
}
return true;
return true;
}
private boolean internalQueueTincrement(TIncrement inc) throws TException {
byte[][] famAndQf = CellUtil.parseColumn(inc.getColumn());
if (famAndQf.length != 2) return false;
if (famAndQf.length != 2) {
return false;
}
return internalQueueIncrement(inc.getTable(), inc.getRow(), famAndQf[0], famAndQf[1],
inc.getAmmount());
@ -207,7 +232,6 @@ public class IncrementCoalescer implements IncrementCoalescerMBean {
byte[] qual, long ammount) throws TException {
int countersMapSize = countersMap.size();
//Make sure that the number of threads is scaled.
dynamicallySetCoreSize(countersMapSize);
@ -293,7 +317,7 @@ public class IncrementCoalescer implements IncrementCoalescerMBean {
/**
* This method samples the incoming requests and, if selected, will check if
* the corePoolSize should be changed.
* @param countersMapSize
* @param countersMapSize the size of the counters map
*/
private void dynamicallySetCoreSize(int countersMapSize) {
// Here we are using countersMapSize as a random number, meaning this
@ -302,9 +326,10 @@ public class IncrementCoalescer implements IncrementCoalescerMBean {
return;
}
double currentRatio = (double) countersMapSize / (double) maxQueueSize;
int newValue = 1;
int newValue;
if (currentRatio < 0.1) {
// it's 1
newValue = 1;
} else if (currentRatio < 0.3) {
newValue = 2;
} else if (currentRatio < 0.5) {
@ -316,6 +341,7 @@ public class IncrementCoalescer implements IncrementCoalescerMBean {
} else {
newValue = 22;
}
if (pool.getCorePoolSize() != newValue) {
pool.setCorePoolSize(newValue);
}

2
hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift/TBoundedThreadPoolServer.java
View File

@ -258,7 +258,7 @@ public class TBoundedThreadPoolServer extends TServer {
serverTransport_.interrupt();
}
private class ClientConnnection implements Runnable {
private final class ClientConnnection implements Runnable {
private TTransport client;

6
hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift/ThriftMetrics.java
View File

@ -67,9 +67,11 @@ public class ThriftMetrics {
slowResponseTime = conf.getLong(SLOW_RESPONSE_NANO_SEC, DEFAULT_SLOW_RESPONSE_NANO_SEC);
if (t == ThriftServerType.ONE) {
source = CompatibilitySingletonFactory.getInstance(MetricsThriftServerSourceFactory.class).createThriftOneSource();
source = CompatibilitySingletonFactory.getInstance(MetricsThriftServerSourceFactory.class)
.createThriftOneSource();
} else if (t == ThriftServerType.TWO) {
source = CompatibilitySingletonFactory.getInstance(MetricsThriftServerSourceFactory.class).createThriftTwoSource();
source = CompatibilitySingletonFactory.getInstance(MetricsThriftServerSourceFactory.class)
.createThriftTwoSource();
}
}

9
hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift/ThriftServer.java
View File

@ -87,15 +87,15 @@ public class ThriftServer {
/**
* Start up or shuts down the Thrift server, depending on the arguments.
* @param args
* @param args the arguments to pass in when starting the Thrift server
*/
void doMain(final String[] args) throws Exception {
processOptions(args);
serverRunner = new ThriftServerRunner(conf);
// Put up info server.
int port = conf.getInt("hbase.thrift.info.port", 9095);
if (port >= 0) {
conf.setLong("startcode", System.currentTimeMillis());
String a = conf.get("hbase.thrift.info.bindAddress", "0.0.0.0");
@ -103,6 +103,7 @@ public class ThriftServer {
infoServer.setAttribute("hbase.conf", conf);
infoServer.start();
}
serverRunner.run();
}
@ -230,10 +231,6 @@ public class ThriftServer {
}
}
/**
* @param args
* @throws Exception
*/
public static void main(String [] args) throws Exception {
LOG.info("***** STARTING service '" + ThriftServer.class.getSimpleName() + "' *****");
VersionInfo.logVersion();

86
hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift/ThriftServerRunner.java
View File

@ -161,7 +161,8 @@ public class ThriftServerRunner implements Runnable {
static final String BIND_CONF_KEY = "hbase.regionserver.thrift.ipaddress";
static final String COMPACT_CONF_KEY = "hbase.regionserver.thrift.compact";
static final String FRAMED_CONF_KEY = "hbase.regionserver.thrift.framed";
static final String MAX_FRAME_SIZE_CONF_KEY = "hbase.regionserver.thrift.framed.max_frame_size_in_mb";
static final String MAX_FRAME_SIZE_CONF_KEY =
"hbase.regionserver.thrift.framed.max_frame_size_in_mb";
static final String PORT_CONF_KEY = "hbase.regionserver.thrift.port";
static final String COALESCE_INC_KEY = "hbase.regionserver.thrift.coalesceIncrement";
static final String USE_HTTP_CONF_KEY = "hbase.regionserver.thrift.http";
@ -347,7 +348,8 @@ public class ThriftServerRunner implements Runnable {
doAsEnabled = conf.getBoolean(THRIFT_SUPPORT_PROXYUSER, false);
if (doAsEnabled) {
if (!conf.getBoolean(USE_HTTP_CONF_KEY, false)) {
LOG.warn("Fail to enable the doAs feature. hbase.regionserver.thrift.http is not configured ");
LOG.warn("Fail to enable the doAs feature. hbase.regionserver.thrift.http is not " +
"configured ");
}
}
if (qop != null) {
@ -433,7 +435,8 @@ public class ThriftServerRunner implements Runnable {
httpServer = new Server(threadPool);
// Context handler
ServletContextHandler ctxHandler = new ServletContextHandler(httpServer, "/", ServletContextHandler.SESSIONS);
ServletContextHandler ctxHandler = new ServletContextHandler(httpServer, "/",
ServletContextHandler.SESSIONS);
ctxHandler.addServlet(new ServletHolder(thriftHttpServlet), "/*");
// set up Jetty and run the embedded server
@ -508,14 +511,7 @@ public class ThriftServerRunner implements Runnable {
*/
private void setupServer() throws Exception {
// Construct correct ProtocolFactory
TProtocolFactory protocolFactory;
if (conf.getBoolean(COMPACT_CONF_KEY, false)) {
LOG.debug("Using compact protocol");
protocolFactory = new TCompactProtocol.Factory();
} else {
LOG.debug("Using binary protocol");
protocolFactory = new TBinaryProtocol.Factory();
}
TProtocolFactory protocolFactory = getProtocolFactory();
final TProcessor p = new Hbase.Processor<>(handler);
ImplType implType = ImplType.getServerImpl(conf);
@ -614,10 +610,8 @@ public class ThriftServerRunner implements Runnable {
CallQueue callQueue = new CallQueue(new LinkedBlockingQueue<>(), metrics);
ExecutorService executorService = createExecutor(
callQueue, serverArgs.getMaxWorkerThreads(), serverArgs.getMaxWorkerThreads());
serverArgs.executorService(executorService)
.processor(processor)
.transportFactory(transportFactory)
.protocolFactory(protocolFactory);
serverArgs.executorService(executorService).processor(processor)
.transportFactory(transportFactory).protocolFactory(protocolFactory);
tserver = new THsHaServer(serverArgs);
} else { // THREADED_SELECTOR
TThreadedSelectorServer.Args serverArgs =
@ -625,10 +619,8 @@ public class ThriftServerRunner implements Runnable {
CallQueue callQueue = new CallQueue(new LinkedBlockingQueue<>(), metrics);
ExecutorService executorService = createExecutor(
callQueue, serverArgs.getWorkerThreads(), serverArgs.getWorkerThreads());
serverArgs.executorService(executorService)
.processor(processor)
.transportFactory(transportFactory)
.protocolFactory(protocolFactory);
serverArgs.executorService(executorService).processor(processor)
.transportFactory(transportFactory).protocolFactory(protocolFactory);
tserver = new TThreadedSelectorServer(serverArgs);
}
LOG.info("starting HBase " + implType.simpleClassName() +
@ -640,21 +632,17 @@ public class ThriftServerRunner implements Runnable {
THRIFT_SERVER_SOCKET_READ_TIMEOUT_DEFAULT);
TServerTransport serverTransport = new TServerSocket(
new TServerSocket.ServerSocketTransportArgs().
bindAddr(new InetSocketAddress(listenAddress, listenPort)).
backlog(backlog).
bindAddr(new InetSocketAddress(listenAddress, listenPort)).backlog(backlog).
clientTimeout(readTimeout));
TBoundedThreadPoolServer.Args serverArgs =
new TBoundedThreadPoolServer.Args(serverTransport, conf);
serverArgs.processor(processor)
.transportFactory(transportFactory)
serverArgs.processor(processor).transportFactory(transportFactory)
.protocolFactory(protocolFactory);
LOG.info("starting " + ImplType.THREAD_POOL.simpleClassName() + " on "
+ listenAddress + ":" + Integer.toString(listenPort)
+ " with readTimeout " + readTimeout + "ms; " + serverArgs);
TBoundedThreadPoolServer tserver =
new TBoundedThreadPoolServer(serverArgs, metrics);
this.tserver = tserver;
this.tserver = new TBoundedThreadPoolServer(serverArgs, metrics);
} else {
throw new AssertionError("Unsupported Thrift server implementation: " +
implType.simpleClassName());
@ -672,6 +660,20 @@ public class ThriftServerRunner implements Runnable {
registerFilters(conf);
}
private TProtocolFactory getProtocolFactory() {
TProtocolFactory protocolFactory;
if (conf.getBoolean(COMPACT_CONF_KEY, false)) {
LOG.debug("Using compact protocol");
protocolFactory = new TCompactProtocol.Factory();
} else {
LOG.debug("Using binary protocol");
protocolFactory = new TBinaryProtocol.Factory();
}
return protocolFactory;
}
ExecutorService createExecutor(BlockingQueue<Runnable> callQueue,
int minWorkers, int maxWorkers) {
ThreadFactoryBuilder tfb = new ThreadFactoryBuilder();
@ -749,10 +751,9 @@ public class ThriftServerRunner implements Runnable {
* @param tableName
* name of table
* @return Table object
* @throws IOException
* @throws IOException if getting the table fails
*/
public Table getTable(final byte[] tableName) throws
IOException {
public Table getTable(final byte[] tableName) throws IOException {
String table = Bytes.toString(tableName);
return connectionCache.getTable(table);
}
@ -765,7 +766,7 @@ public class ThriftServerRunner implements Runnable {
* Assigns a unique ID to the scanner and adds the mapping to an internal
* hash-map.
*
* @param scanner
* @param scanner the {@link ResultScanner} to add
* @return integer scanner id
*/
protected synchronized int addScanner(ResultScanner scanner, boolean sortColumns) {
@ -778,7 +779,7 @@ public class ThriftServerRunner implements Runnable {
/**
* Returns the scanner associated with the specified ID.
*
* @param id
* @param id the ID of the scanner to get
* @return a Scanner, or null if ID was invalid.
*/
protected synchronized ResultScannerWrapper getScanner(int id) {
@ -789,7 +790,7 @@ public class ThriftServerRunner implements Runnable {
* Removes the scanner associated with the specified ID from the internal
* id-&gt;scanner hash-map.
*
* @param id
* @param id the ID of the scanner to remove
* @return a Scanner, or null if ID was invalid.
*/
protected synchronized ResultScannerWrapper removeScanner(int id) {
@ -1361,10 +1362,12 @@ public class ThriftServerRunner implements Runnable {
put.setDurability(m.writeToWAL ? Durability.SYNC_WAL : Durability.SKIP_WAL);
}
}
if (!delete.isEmpty())
if (!delete.isEmpty()) {
table.delete(delete);
if (!put.isEmpty())
}
if (!put.isEmpty()) {
table.put(put);
}
} catch (IOException e) {
LOG.warn(e.getMessage(), e);
throw getIOError(e);
@ -1434,19 +1437,23 @@ public class ThriftServerRunner implements Runnable {
put.setDurability(m.writeToWAL ? Durability.SYNC_WAL : Durability.SKIP_WAL);
}
}
if (!delete.isEmpty())
if (!delete.isEmpty()) {
deletes.add(delete);
if (!put.isEmpty())
}
if (!put.isEmpty()) {
puts.add(put);
}
}
Table table = null;
try {
table = getTable(tableName);
if (!puts.isEmpty())
if (!puts.isEmpty()) {
table.put(puts);
if (!deletes.isEmpty())
}
if (!deletes.isEmpty()) {
table.delete(deletes);
}
} catch (IOException e) {
LOG.warn(e.getMessage(), e);
@ -1762,8 +1769,7 @@ public class ThriftServerRunner implements Runnable {
}
}
private void closeTable(Table table) throws IOError
{
private void closeTable(Table table) throws IOError {
try{
if(table != null){
table.close();

19
hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift/ThriftUtilities.java
View File

@ -46,16 +46,17 @@ import org.apache.hadoop.hbase.util.Bytes;
import org.apache.yetus.audience.InterfaceAudience;
@InterfaceAudience.Private
public class ThriftUtilities {
public final class ThriftUtilities {
private ThriftUtilities() {
}
/**
* This utility method creates a new Hbase HColumnDescriptor object based on a
* Thrift ColumnDescriptor "struct".
*
* @param in
* Thrift ColumnDescriptor object
* @param in Thrift ColumnDescriptor object
* @return HColumnDescriptor
* @throws IllegalArgument
* @throws IllegalArgument if the column name is empty
*/
static public HColumnDescriptor colDescFromThrift(ColumnDescriptor in)
throws IllegalArgument {
@ -154,8 +155,10 @@ public class ThriftUtilities {
if(result_ == null || result_.isEmpty()) {
continue;
}
TRowResult result = new TRowResult();
result.row = ByteBuffer.wrap(result_.getRow());
if (sortColumns) {
result.sortedColumns = new ArrayList<>();
for (Cell kv : result_.rawCells()) {
@ -173,8 +176,10 @@ public class ThriftUtilities {
new TCell(ByteBuffer.wrap(CellUtil.cloneValue(kv)), kv.getTimestamp()));
}
}
results.add(result);
}
return results;
}
@ -204,7 +209,11 @@ public class ThriftUtilities {
public static Increment incrementFromThrift(TIncrement tincrement) {
Increment inc = new Increment(tincrement.getRow());
byte[][] famAndQf = CellUtil.parseColumn(tincrement.getColumn());
if (famAndQf.length != 2) return null;
if (famAndQf.length != 2) {
return null;
}
inc.addColumn(famAndQf[0], famAndQf[1], tincrement.getAmmount());
return inc;
}

232
hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift2/ThriftServer.java
View File

@ -89,8 +89,8 @@ import org.slf4j.LoggerFactory;
import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
/**
* ThriftServer - this class starts up a Thrift server which implements the HBase API specified in the
* HbaseClient.thrift IDL file.
* ThriftServer - this class starts up a Thrift server which implements the HBase API specified in
* the HbaseClient.thrift IDL file.
*/
@InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.TOOLS)
@SuppressWarnings({ "rawtypes", "unchecked" })
@ -155,11 +155,14 @@ public class ThriftServer extends Configured implements Tool {
options.addOption("ro", "readonly", false,
"Respond only to read method requests [default: false]");
OptionGroup servers = new OptionGroup();
servers.addOption(
new Option("nonblocking", false, "Use the TNonblockingServer. This implies the framed transport."));
servers.addOption(new Option("hsha", false, "Use the THsHaServer. This implies the framed transport."));
servers.addOption(new Option("selector", false, "Use the TThreadedSelectorServer. This implies the framed transport."));
servers.addOption(new Option("threadpool", false, "Use the TThreadPoolServer. This is the default."));
servers.addOption(new Option("nonblocking", false,
"Use the TNonblockingServer. This implies the framed transport."));
servers.addOption(new Option("hsha", false,
"Use the THsHaServer. This implies the framed transport."));
servers.addOption(new Option("selector", false,
"Use the TThreadedSelectorServer. This implies the framed transport."));
servers.addOption(new Option("threadpool", false,
"Use the TThreadPoolServer. This is the default."));
options.addOptionGroup(servers);
return options;
}
@ -243,8 +246,9 @@ public class ThriftServer extends Configured implements Tool {
}
}
private static TServer getTNonBlockingServer(TProtocolFactory protocolFactory, TProcessor processor,
TTransportFactory transportFactory, InetSocketAddress inetSocketAddress) throws TTransportException {
private static TServer getTNonBlockingServer(TProtocolFactory protocolFactory,
TProcessor processor, TTransportFactory transportFactory, InetSocketAddress inetSocketAddress)
throws TTransportException {
TNonblockingServerTransport serverTransport = new TNonblockingServerSocket(inetSocketAddress);
log.info("starting HBase Nonblocking Thrift server on " + inetSocketAddress.toString());
TNonblockingServer.Args serverArgs = new TNonblockingServer.Args(serverTransport);
@ -378,31 +382,18 @@ public class ThriftServer extends Configured implements Tool {
@Override
public int run(String[] args) throws Exception {
final Configuration conf = getConf();
TServer server = null;
Options options = getOptions();
CommandLine cmd = parseArguments(conf, options, args);
int workerThreads = 0;
int selectorThreads = 0;
int maxCallQueueSize = -1; // use unbounded queue by default
/**
* This is to please both bin/hbase and bin/hbase-daemon. hbase-daemon provides "start" and "stop" arguments hbase
* should print the help if no argument is provided
*/
List<?> argList = cmd.getArgList();
if (cmd.hasOption("help") || !argList.contains("start") || argList.contains("stop")) {
printUsage();
if (checkArguments(cmd)) {
return 1;
}
// Get address to bind
String bindAddress;
if (cmd.hasOption("bind")) {
bindAddress = cmd.getOptionValue("bind");
conf.set("hbase.thrift.info.bindAddress", bindAddress);
} else {
bindAddress = conf.get("hbase.thrift.info.bindAddress");
}
String bindAddress = getBindAddress(conf, cmd);
// check if server should only process read requests, if so override the conf
if (cmd.hasOption("readonly")) {
@ -413,35 +404,13 @@ public class ThriftServer extends Configured implements Tool {
}
// Get read timeout
int readTimeout = THRIFT_SERVER_SOCKET_READ_TIMEOUT_DEFAULT;
if (cmd.hasOption(READ_TIMEOUT_OPTION)) {
try {
readTimeout = Integer.parseInt(cmd.getOptionValue(READ_TIMEOUT_OPTION));
} catch (NumberFormatException e) {
throw new RuntimeException("Could not parse the value provided for the timeout option", e);
}
} else {
readTimeout = conf.getInt(THRIFT_SERVER_SOCKET_READ_TIMEOUT_KEY,
THRIFT_SERVER_SOCKET_READ_TIMEOUT_DEFAULT);
}
int readTimeout = getReadTimeout(conf, cmd);
// Get port to bind to
int listenPort = 0;
try {
if (cmd.hasOption("port")) {
listenPort = Integer.parseInt(cmd.getOptionValue("port"));
} else {
listenPort = conf.getInt("hbase.regionserver.thrift.port", DEFAULT_LISTEN_PORT);
}
} catch (NumberFormatException e) {
throw new RuntimeException("Could not parse the value provided for the port option", e);
}
int listenPort = getListenPort(conf, cmd);
// Thrift's implementation uses '0' as a placeholder for 'use the default.'
int backlog = conf.getInt(BACKLOG_CONF_KEY, 0);
// Local hostname and user name,
// used only if QOP is configured.
// Local hostname and user name, used only if QOP is configured.
String host = null;
String name = null;
@ -453,8 +422,7 @@ public class ThriftServer extends Configured implements Tool {
host = Strings.domainNamePointerToHostName(DNS.getDefaultHost(
conf.get("hbase.thrift.dns.interface", "default"),
conf.get("hbase.thrift.dns.nameserver", "default")));
userProvider.login("hbase.thrift.keytab.file",
"hbase.thrift.kerberos.principal", host);
userProvider.login("hbase.thrift.keytab.file", "hbase.thrift.kerberos.principal", host);
}
UserGroupInformation realUser = userProvider.getCurrent().getUGI();
@ -463,12 +431,10 @@ public class ThriftServer extends Configured implements Tool {
if (stringQop != null) {
qop = SaslUtil.getQop(stringQop);
if (!securityEnabled) {
throw new IOException("Thrift server must"
+ " run in secure mode to support authentication");
throw new IOException("Thrift server must run in secure mode to support authentication");
}
// Extract the name from the principal
name = SecurityUtil.getUserFromPrincipal(
conf.get("hbase.thrift.kerberos.principal"));
name = SecurityUtil.getUserFromPrincipal(conf.get("hbase.thrift.kerberos.principal"));
}
boolean nonblocking = cmd.hasOption("nonblocking");
@ -478,14 +444,7 @@ public class ThriftServer extends Configured implements Tool {
ThriftMetrics metrics = new ThriftMetrics(conf, ThriftMetrics.ThriftServerType.TWO);
final JvmPauseMonitor pauseMonitor = new JvmPauseMonitor(conf, metrics.getSource());
String implType = "threadpool";
if (nonblocking) {
implType = "nonblocking";
} else if (hsha) {
implType = "hsha";
} else if (selector) {
implType = "selector";
}
String implType = getImplType(nonblocking, hsha, selector);
conf.set("hbase.regionserver.thrift.server.type", implType);
conf.setInt("hbase.regionserver.thrift.port", listenPort);
@ -549,49 +508,12 @@ public class ThriftServer extends Configured implements Tool {
}
// Put up info server.
int port = conf.getInt("hbase.thrift.info.port", 9095);
if (port >= 0) {
conf.setLong("startcode", System.currentTimeMillis());
String a = conf.get("hbase.thrift.info.bindAddress", "0.0.0.0");
InfoServer infoServer = new InfoServer("thrift", a, port, false, conf);
infoServer.setAttribute("hbase.conf", conf);
infoServer.start();
}
startInfoServer(conf);
if (nonblocking) {
server = getTNonBlockingServer(protocolFactory,
processor,
transportFactory,
inetSocketAddress);
} else if (hsha) {
server = getTHsHaServer(protocolFactory,
processor,
transportFactory,
workerThreads,
maxCallQueueSize,
inetSocketAddress,
metrics);
} else if (selector) {
server = getTThreadedSelectorServer(protocolFactory,
processor,
transportFactory,
workerThreads,
selectorThreads,
maxCallQueueSize,
inetSocketAddress,
metrics);
} else {
server = getTThreadPoolServer(protocolFactory,
processor,
transportFactory,
workerThreads,
inetSocketAddress,
backlog,
readTimeout,
metrics);
}
final TServer tserver = getServer(workerThreads, selectorThreads, maxCallQueueSize, readTimeout,
backlog, nonblocking, hsha, selector, metrics, protocolFactory, processor,
transportFactory, inetSocketAddress);
final TServer tserver = server;
realUser.doAs(
new PrivilegedAction<Object>() {
@Override
@ -608,4 +530,106 @@ public class ThriftServer extends Configured implements Tool {
// when tserver.stop eventually happens we'll get here.
return 0;
}
private String getImplType(boolean nonblocking, boolean hsha, boolean selector) {
String implType = "threadpool";
if (nonblocking) {
implType = "nonblocking";
} else if (hsha) {
implType = "hsha";
} else if (selector) {
implType = "selector";
}
return implType;
}
private boolean checkArguments(CommandLine cmd) {
/*
* This is to please both bin/hbase and bin/hbase-daemon. hbase-daemon provides "start" and
* "stop" arguments hbase should print the help if no argument is provided
*/
List<?> argList = cmd.getArgList();
if (cmd.hasOption("help") || !argList.contains("start") || argList.contains("stop")) {
printUsage();
return true;
}
return false;
}
private String getBindAddress(Configuration conf, CommandLine cmd) {
String bindAddress;
if (cmd.hasOption("bind")) {
bindAddress = cmd.getOptionValue("bind");
conf.set("hbase.thrift.info.bindAddress", bindAddress);
} else {
bindAddress = conf.get("hbase.thrift.info.bindAddress");
}
return bindAddress;
}
private int getListenPort(Configuration conf, CommandLine cmd) {
int listenPort;
try {
if (cmd.hasOption("port")) {
listenPort = Integer.parseInt(cmd.getOptionValue("port"));
} else {
listenPort = conf.getInt("hbase.regionserver.thrift.port", DEFAULT_LISTEN_PORT);
}
} catch (NumberFormatException e) {
throw new RuntimeException("Could not parse the value provided for the port option", e);
}
return listenPort;
}
private int getReadTimeout(Configuration conf, CommandLine cmd) {
int readTimeout;
if (cmd.hasOption(READ_TIMEOUT_OPTION)) {
try {
readTimeout = Integer.parseInt(cmd.getOptionValue(READ_TIMEOUT_OPTION));
} catch (NumberFormatException e) {
throw new RuntimeException("Could not parse the value provided for the timeout option", e);
}
} else {
readTimeout = conf.getInt(THRIFT_SERVER_SOCKET_READ_TIMEOUT_KEY,
THRIFT_SERVER_SOCKET_READ_TIMEOUT_DEFAULT);
}
return readTimeout;
}
private void startInfoServer(Configuration conf) throws IOException {
int port = conf.getInt("hbase.thrift.info.port", 9095);
if (port >= 0) {
conf.setLong("startcode", System.currentTimeMillis());
String a = conf.get("hbase.thrift.info.bindAddress", "0.0.0.0");
InfoServer infoServer = new InfoServer("thrift", a, port, false, conf);
infoServer.setAttribute("hbase.conf", conf);
infoServer.start();
}
}
private TServer getServer(int workerThreads, int selectorThreads, int maxCallQueueSize,
int readTimeout, int backlog, boolean nonblocking, boolean hsha, boolean selector,
ThriftMetrics metrics, TProtocolFactory protocolFactory, TProcessor processor,
TTransportFactory transportFactory, InetSocketAddress inetSocketAddress)
throws TTransportException {
TServer server;
if (nonblocking) {
server = getTNonBlockingServer(protocolFactory, processor, transportFactory,
inetSocketAddress);
} else if (hsha) {
server = getTHsHaServer(protocolFactory, processor, transportFactory, workerThreads,
maxCallQueueSize, inetSocketAddress, metrics);
} else if (selector) {
server = getTThreadedSelectorServer(protocolFactory, processor, transportFactory,
workerThreads, selectorThreads, maxCallQueueSize, inetSocketAddress, metrics);
} else {
server = getTThreadPoolServer(protocolFactory, processor, transportFactory, workerThreads,
inetSocketAddress, backlog, readTimeout, metrics);
}
return server;
}
}

13
hbase-thrift/src/main/java/org/apache/hadoop/hbase/thrift2/ThriftUtilities.java
View File

@ -74,7 +74,7 @@ import org.apache.hadoop.hbase.util.Bytes;
import org.apache.yetus.audience.InterfaceAudience;
@InterfaceAudience.Private
public class ThriftUtilities {
public final class ThriftUtilities {
private ThriftUtilities() {
throw new UnsupportedOperationException("Can't initialize class");
@ -316,6 +316,8 @@ public class ThriftUtilities {
"Timestamp is required for TDelete with DeleteFamilyVersion type");
}
break;
default:
throw new IllegalArgumentException("DeleteType is required for TDelete");
}
} else {
throw new IllegalArgumentException("DeleteType is required for TDelete");
@ -416,12 +418,15 @@ public class ThriftUtilities {
public static Scan scanFromThrift(TScan in) throws IOException {
Scan out = new Scan();
if (in.isSetStartRow())
if (in.isSetStartRow()) {
out.setStartRow(in.getStartRow());
if (in.isSetStopRow())
}
if (in.isSetStopRow()) {
out.setStopRow(in.getStopRow());
if (in.isSetCaching())
}
if (in.isSetCaching()) {
out.setCaching(in.getCaching());
}
if (in.isSetMaxVersions()) {
out.setMaxVersions(in.getMaxVersions());
}

51
hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift/TestThriftServer.java
View File

@ -101,7 +101,7 @@ public class TestThriftServer {
private static ByteBuffer valueBname = asByteBuffer("valueB");
private static ByteBuffer valueCname = asByteBuffer("valueC");
private static ByteBuffer valueDname = asByteBuffer("valueD");
private static ByteBuffer valueEname = asByteBuffer(100l);
private static ByteBuffer valueEname = asByteBuffer(100L);
@Rule
public TestName name = new TestName();
@ -124,8 +124,6 @@ public class TestThriftServer {
* consolidate all testing to one method because HBaseClusterTestCase
* is prone to OutOfMemoryExceptions when there are three or more
* JUnit test methods.
*
* @throws Exception
*/
@Test
public void testAll() throws Exception {
@ -147,8 +145,6 @@ public class TestThriftServer {
* Tests for creating, enabling, disabling, and deleting tables. Also
* tests that creating a table with an invalid column name yields an
* IllegalArgument exception.
*
* @throws Exception
*/
public void doTestTableCreateDrop() throws Exception {
ThriftServerRunner.HBaseHandler handler =
@ -181,10 +177,6 @@ public class TestThriftServer {
* TODO: These counts are supposed to be zero but sometimes they are not, they are equal to the
* passed in maybe. Investigate why. My guess is they are set by the test that runs just
* previous to this one. Sometimes they are cleared. Sometimes not.
* @param name
* @param maybe
* @param metrics
* @return
*/
private int getCurrentCount(final String name, final int maybe, final ThriftMetrics metrics) {
int currentCount = 0;
@ -221,8 +213,10 @@ public class TestThriftServer {
// 3 to 6 seconds (to account for potential slowness), measured in nanoseconds
try {
metricsHelper.assertGaugeGt("getTableNames_avg_time", 3L * 1000 * 1000 * 1000, metrics.getSource());
metricsHelper.assertGaugeLt("getTableNames_avg_time",6L * 1000 * 1000 * 1000, metrics.getSource());
metricsHelper.assertGaugeGt("getTableNames_avg_time", 3L * 1000 * 1000 * 1000,
metrics.getSource());
metricsHelper.assertGaugeLt("getTableNames_avg_time",6L * 1000 * 1000 * 1000,
metrics.getSource());
} catch (AssertionError e) {
LOG.info("Fix me! Why does this happen? A concurrent cluster running?", e);
}
@ -300,9 +294,11 @@ public class TestThriftServer {
Thread.sleep(1000);
long lv = handler.get(tableAname, rowAname, columnAname, null).get(0).value.getLong();
// Wait on all increments being flushed
while (handler.coalescer.getQueueSize() != 0) Threads.sleep(10);
assertEquals((100 + (2 * numIncrements)), lv );
while (handler.coalescer.getQueueSize() != 0) {
Threads.sleep(10);
}
assertEquals((100 + (2 * numIncrements)), lv);
lv = handler.get(tableAname, rowBname, columnAAname, null).get(0).value.getLong();
assertEquals((100 + (3 * 7 * numIncrements)), lv);
@ -315,8 +311,6 @@ public class TestThriftServer {
* Tests adding a series of Mutations and BatchMutations, including a
* delete mutation. Also tests data retrieval, and getting back multiple
* versions.
*
* @throws Exception
*/
public void doTestTableMutations() throws Exception {
ThriftServerRunner.HBaseHandler handler =
@ -393,8 +387,6 @@ public class TestThriftServer {
* Similar to testTableMutations(), except Mutations are applied with
* specific timestamps and data retrieval uses these timestamps to
* extract specific versions of data.
*
* @throws Exception
*/
public void doTestTableTimestampsAndColumns() throws Exception {
// Setup
@ -473,8 +465,6 @@ public class TestThriftServer {
/**
* Tests the four different scanner-opening methods (with and without
* a stoprow, with and without a timestamp).
*
* @throws Exception
*/
public void doTestTableScanners() throws Exception {
// Setup
@ -515,7 +505,8 @@ public class TestThriftServer {
closeScanner(scanner1, handler);
// Test a scanner on all rows and all columns, with timestamp
int scanner2 = handler.scannerOpenTs(tableAname, rowAname, getColumnList(true, true), time1, null);
int scanner2 = handler.scannerOpenTs(tableAname, rowAname, getColumnList(true, true), time1,
null);
TRowResult rowResult2a = handler.scannerGet(scanner2).get(0);
assertEquals(rowResult2a.columns.size(), 1);
// column A deleted, does not exist.
@ -594,8 +585,6 @@ public class TestThriftServer {
/**
* For HBASE-2556
* Tests for GetTableRegions
*
* @throws Exception
*/
public void doTestGetTableRegions() throws Exception {
ThriftServerRunner.HBaseHandler handler =
@ -659,8 +648,6 @@ public class TestThriftServer {
/**
* Appends the value to a cell and checks that the cell value is updated properly.
*
* @throws Exception
*/
public static void doTestAppend() throws Exception {
ThriftServerRunner.HBaseHandler handler =
@ -693,8 +680,6 @@ public class TestThriftServer {
/**
* Check that checkAndPut fails if the cell does not exist, then put in the cell, then check that
* the checkAndPut succeeds.
*
* @throws Exception
*/
public static void doTestCheckAndPut() throws Exception {
ThriftServerRunner.HBaseHandler handler =
@ -791,7 +776,6 @@ public class TestThriftServer {
}
/**
*
* @return a List of ColumnDescriptors for use in creating a table. Has one
* default ColumnDescriptor and one ColumnDescriptor with fewer versions
*/
@ -819,13 +803,18 @@ public class TestThriftServer {
*/
private List<ByteBuffer> getColumnList(boolean includeA, boolean includeB) {
List<ByteBuffer> columnList = new ArrayList<>();
if (includeA) columnList.add(columnAname);
if (includeB) columnList.add(columnBname);
if (includeA) {
columnList.add(columnAname);
}
if (includeB) {
columnList.add(columnBname);
}
return columnList;
}
/**
*
* @return a List of Mutations for a row, with columnA having valueA
* and columnB having valueB
*/
@ -837,7 +826,6 @@ public class TestThriftServer {
}
/**
*
* @return a List of BatchMutations with the following effects:
* (rowA, columnA): delete
* (rowA, columnB): place valueC
@ -871,7 +859,6 @@ public class TestThriftServer {
*
* @param scannerId the scanner to close
* @param handler the HBaseHandler interfacing to HBase
* @throws Exception
*/
private void closeScanner(
int scannerId, ThriftServerRunner.HBaseHandler handler) throws Exception {

5
hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift/TestThriftServerCmdLine.java
View File

@ -89,8 +89,9 @@ public class TestThriftServerCmdLine {
continue;
}
for (boolean specifyCompact : new boolean[] {false, true}) {
parameters.add(new Object[]{implType, specifyFramed,
specifyBindIP, specifyCompact});
parameters.add(new Object[] {
implType, specifyFramed, specifyBindIP, specifyCompact
});
}
}
}

17
hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift2/TestThriftHBaseServiceHandler.java
View File

@ -590,8 +590,6 @@ public class TestThriftHBaseServiceHandler {
/**
* check that checkAndPut fails if the cell does not exist, then put in the cell, then check
* that the checkAndPut succeeds.
*
* @throws Exception
*/
@Test
public void testCheckAndPut() throws Exception {
@ -637,8 +635,6 @@ public class TestThriftHBaseServiceHandler {
/**
* check that checkAndDelete fails if the cell does not exist, then put in the cell, then
* check that the checkAndDelete succeeds.
*
* @throws Exception
*/
@Test
public void testCheckAndDelete() throws Exception {
@ -733,8 +729,7 @@ public class TestThriftHBaseServiceHandler {
/**
* Tests keeping a HBase scanner alive for long periods of time. Each call to getScannerRow()
* should reset the ConnectionCache timeout for the scanner's connection
* @throws Exception
* should reset the ConnectionCache timeout for the scanner's connection.
*/
@Test
public void testLongLivedScan() throws Exception {
@ -1047,7 +1042,11 @@ public class TestThriftHBaseServiceHandler {
*/
private String pad(int n, byte pad) {
String res = Integer.toString(n);
while (res.length() < pad) res = "0" + res;
while (res.length() < pad) {
res = "0" + res;
}
return res;
}
@ -1381,8 +1380,6 @@ public class TestThriftHBaseServiceHandler {
/**
* Put valueA to a row, make sure put has happened, then create a mutation object to put valueB
* and delete ValueA, then check that the row value is only valueB.
*
* @throws Exception
*/
@Test
public void testMutateRow() throws Exception {
@ -1445,8 +1442,6 @@ public class TestThriftHBaseServiceHandler {
* Create TPut, TDelete , TIncrement objects, set durability then call ThriftUtility
* functions to get Put , Delete and Increment respectively. Use getDurability to make sure
* the returned objects have the appropriate durability setting.
*
* @throws Exception
*/
@Test
public void testDurability() throws Exception {

5
hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift2/TestThriftHBaseServiceHandlerWithLabels.java
View File

@ -459,8 +459,11 @@ public void testAppend() throws Exception {
*/
private String pad(int n, byte pad) {
String res = Integer.toString(n);
while (res.length() < pad)
while (res.length() < pad) {
res = "0" + res;
}
return res;
}
}