diff --git a/docs/interpreter/hbase.md b/docs/interpreter/hbase.md index 0afbc726fab..304ce91124b 100644 --- a/docs/interpreter/hbase.md +++ b/docs/interpreter/hbase.md @@ -39,10 +39,15 @@ Zeppelin is built against HBase 1.x and 2.x releases. Description - hbase.home + HBASE_HOME /usr/lib/hbase Installation directory of HBase, defaults to HBASE_HOME in environment + + shell.command.timeout.millisecs + 600000 + HBase shell command timeout in millisecs when there is no output, defaults to 10 minutes + If you want to connect to HBase running on a cluster, you'll need to follow the next step. diff --git a/hbase/pom.xml b/hbase/pom.xml index d2f9436609b..9740ee3ea3b 100644 --- a/hbase/pom.xml +++ b/hbase/pom.xml @@ -37,12 +37,9 @@ - commons-io - commons-io - - - org.apache.commons - commons-lang3 + org.apache.zeppelin + zeppelin-shell + ${project.version} diff --git a/hbase/src/main/java/org/apache/zeppelin/hbase/HbaseInterpreter.java b/hbase/src/main/java/org/apache/zeppelin/hbase/HbaseInterpreter.java index fd8952ba654..ab46e99fdfa 100644 --- a/hbase/src/main/java/org/apache/zeppelin/hbase/HbaseInterpreter.java +++ b/hbase/src/main/java/org/apache/zeppelin/hbase/HbaseInterpreter.java @@ -14,173 +14,72 @@ package org.apache.zeppelin.hbase; -import java.io.ByteArrayOutputStream; -import java.io.File; -import java.io.IOException; -import java.nio.file.Files; -import java.nio.file.Path; -import java.nio.file.Paths; -import java.util.HashMap; -import java.util.Map; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.util.Properties; - -import org.apache.commons.exec.CommandLine; -import org.apache.commons.exec.DefaultExecutor; -import org.apache.commons.exec.ExecuteException; -import org.apache.commons.exec.ExecuteWatchdog; -import org.apache.commons.exec.Executor; -import org.apache.commons.exec.PumpStreamHandler; -import org.apache.commons.io.FileUtils; -import org.apache.commons.lang3.StringUtils; -import org.apache.zeppelin.interpreter.Interpreter; +import java.util.concurrent.TimeUnit; import org.apache.zeppelin.interpreter.InterpreterContext; -import org.apache.zeppelin.interpreter.InterpreterException; import org.apache.zeppelin.interpreter.InterpreterResult; -import org.apache.zeppelin.scheduler.Scheduler; -import org.apache.zeppelin.scheduler.SchedulerFactory; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * HBase interpreter. It uses the hbase shell to interpret the commands. - */ -public class HbaseInterpreter extends Interpreter { - private static final Logger LOGGER = LoggerFactory.getLogger(HbaseInterpreter.class); - - public static final String HBASE_HOME = "hbase.home"; - - private static final Path TEMP_FOLDER = Paths.get(System.getProperty("java.io.tmpdir"), - "zeppelin-hbase-scripts"); - - private Map runningProcesses = new HashMap<>(); +import org.apache.zeppelin.shell.ShellInterpreter; - private Map tempFiles = new HashMap<>(); +public class HbaseInterpreter extends ShellInterpreter { - private static final int SIGTERM_CODE = 143; - - private long commandTimeout = 60000; - - public HbaseInterpreter(Properties properties) { - super(properties); - } + private static final Logger LOGGER = LoggerFactory.getLogger(HbaseInterpreter.class); - @Override - public void open() throws InterpreterException { - // Do nothing + private static final String HBASE_HOME = "HBASE_HOME"; + private static final String LEGACY_HBASE_HOME = "hbase.home"; + private static final String DEFAULT_HBASE_HOME = "/usr/lib/hbase"; + private static final String SHELL_COMMAND_TIMEOUT = "shell.command.timeout.millisecs"; + // Bound the inactivity timeout so that long-running HBase commands are not + // killed while a hung process is still eventually cleaned up. + private static final long DEFAULT_COMMAND_TIMEOUT_MS = TimeUnit.MINUTES.toMillis(10); + + public HbaseInterpreter(Properties property) { + super(property); + if (isBlank(getProperty(SHELL_COMMAND_TIMEOUT))) { + setProperty(SHELL_COMMAND_TIMEOUT, String.valueOf(DEFAULT_COMMAND_TIMEOUT_MS)); + } } @Override - public void close() { - runningProcesses.clear(); - runningProcesses = null; - tempFiles.clear(); - tempFiles = null; + public void open() { + super.open(); + LOGGER.info("HBASE_HOME: {}", getHbaseHome()); } @Override - public InterpreterResult interpret(String st, InterpreterContext context) { - LOGGER.debug("Run HBase shell script: {}", st); - - if (StringUtils.isEmpty(st)) { + public InterpreterResult internalInterpret(String cmd, InterpreterContext context) { + if (isBlank(cmd)) { return new InterpreterResult(InterpreterResult.Code.SUCCESS); } - - String paragraphId = context.getParagraphId(); - final File scriptFile; - try { - // Write script in a temporary file - // The script is enriched with extensions - scriptFile = createTempFile(paragraphId); - FileUtils.write(scriptFile, st + "\nexit"); - } catch (IOException e) { - LOGGER.error("Can not write script in temp file", e); - return new InterpreterResult(InterpreterResult.Code.ERROR, e.getMessage()); - } - - InterpreterResult result = new InterpreterResult(InterpreterResult.Code.SUCCESS); - - final DefaultExecutor executor = new DefaultExecutor(); - final ByteArrayOutputStream errorStream = new ByteArrayOutputStream(); - - executor.setStreamHandler(new PumpStreamHandler(context.out, errorStream)); - executor.setWatchdog(new ExecuteWatchdog(commandTimeout)); - - String hbaseCmdPath = Paths.get(getProperty(HBASE_HOME), "bin", "hbase").toString(); - final CommandLine cmdLine = CommandLine.parse(hbaseCmdPath); - cmdLine.addArgument("shell", false); - cmdLine.addArgument(scriptFile.getAbsolutePath(), false); - - try { - executor.execute(cmdLine); - runningProcesses.put(paragraphId, executor); - } catch (ExecuteException e) { - LOGGER.error("Can not run script in paragraph {}", paragraphId, e); - - final int exitValue = e.getExitValue(); - InterpreterResult.Code code = InterpreterResult.Code.ERROR; - String msg = errorStream.toString(); - - if (exitValue == SIGTERM_CODE) { - code = InterpreterResult.Code.INCOMPLETE; - msg = msg + "Paragraph received a SIGTERM.\n"; - LOGGER.info("The paragraph {} stopped executing: {}", paragraphId, msg); - } - - msg += "ExitValue: " + exitValue; - result = new InterpreterResult(code, msg); - } catch (IOException e) { - LOGGER.error("Can not run script in paragraph {}", paragraphId, e); - result = new InterpreterResult(InterpreterResult.Code.ERROR, e.getMessage()); - } finally { - deleteTempFile(paragraphId); - stopProcess(paragraphId); - } - return result; + String hbaseCommand = buildCommand(cmd); + LOGGER.debug("HBase command: {}", hbaseCommand); + return super.internalInterpret(hbaseCommand, context); } - @Override - public void cancel(InterpreterContext context) { - stopProcess(context.getParagraphId()); - deleteTempFile(context.getParagraphId()); + String buildCommand(String cmd) { + // Quote the command and the hbase path so that quotes, '$', backticks and + // spaces are passed to the hbase shell verbatim. + return "printf '%s\\n' " + shellQuote(cmd.trim()) + + " | " + shellQuote(getHbaseHome() + "/bin/hbase") + " shell -n"; } - @Override - public FormType getFormType() { - return FormType.SIMPLE; - } - - @Override - public int getProgress(InterpreterContext context) { - return 0; - } - - @Override - public Scheduler getScheduler() { - return SchedulerFactory.singleton().createOrGetFIFOScheduler( - HbaseInterpreter.class.getName() + this.hashCode()); - } - - private void stopProcess(String paragraphId) { - Executor executor = runningProcesses.remove(paragraphId); - if (null != executor) { - final ExecuteWatchdog watchdog = executor.getWatchdog(); - watchdog.destroyProcess(); + String getHbaseHome() { + String home = getProperty(HBASE_HOME); + if (isBlank(home)) { + home = getProperty(LEGACY_HBASE_HOME); + } + if (isBlank(home)) { + home = DEFAULT_HBASE_HOME; } + return home; } - private File createTempFile(String paragraphId) throws IOException { - if (!Files.exists(TEMP_FOLDER)) { - Files.createDirectory(TEMP_FOLDER); - } - File temp = Files.createTempFile(TEMP_FOLDER, paragraphId, ".txt").toFile(); - tempFiles.put(paragraphId, temp); - return temp; + static String shellQuote(String value) { + return "'" + value.replace("'", "'\\''") + "'"; } - private void deleteTempFile(String paragraphId) { - File tmpFile = tempFiles.remove(paragraphId); - if (null != tmpFile) { - FileUtils.deleteQuietly(tmpFile); - } + private static boolean isBlank(String value) { + return value == null || value.trim().isEmpty(); } } diff --git a/hbase/src/main/resources/interpreter-setting.json b/hbase/src/main/resources/interpreter-setting.json index 6d7a79c185d..dfaaf338c72 100644 --- a/hbase/src/main/resources/interpreter-setting.json +++ b/hbase/src/main/resources/interpreter-setting.json @@ -4,12 +4,18 @@ "name": "hbase", "className": "org.apache.zeppelin.hbase.HbaseInterpreter", "properties": { - "hbase.home": { + "HBASE_HOME": { "envName": "HBASE_HOME", - "propertyName": "hbase.home", - "defaultValue": "/usr/lib/hbase/", + "propertyName": "HBASE_HOME", + "defaultValue": "/usr/lib/hbase", "description": "Installation directory of HBase", "type": "string" + }, + "shell.command.timeout.millisecs": { + "propertyName": "shell.command.timeout.millisecs", + "defaultValue": "600000", + "description": "HBase shell command timeout in millisecs when there is no output. Default = 600000 (10 minutes)", + "type": "number" } }, "editor": { diff --git a/hbase/src/test/java/org/apache/zeppelin/hbase/HbaseInterpreterTest.java b/hbase/src/test/java/org/apache/zeppelin/hbase/HbaseInterpreterTest.java index 460fc1f0619..7db4903da21 100644 --- a/hbase/src/test/java/org/apache/zeppelin/hbase/HbaseInterpreterTest.java +++ b/hbase/src/test/java/org/apache/zeppelin/hbase/HbaseInterpreterTest.java @@ -14,13 +14,13 @@ package org.apache.zeppelin.hbase; -import org.apache.zeppelin.interpreter.InterpreterException; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; - import java.util.Properties; - -import static org.junit.jupiter.api.Assertions.assertNotNull; +import org.apache.zeppelin.interpreter.InterpreterContext; +import org.apache.zeppelin.interpreter.InterpreterResult; /** * Tests for HBase Interpreter. @@ -29,9 +29,9 @@ public class HbaseInterpreterTest { private static HbaseInterpreter hbaseInterpreter; @BeforeAll - public static void setUp() throws NullPointerException, InterpreterException { + public static void setUp() { Properties properties = new Properties(); - properties.put("hbase.home", ""); + properties.put("HBASE_HOME", ""); hbaseInterpreter = new HbaseInterpreter(properties); hbaseInterpreter.open(); @@ -41,4 +41,92 @@ public static void setUp() throws NullPointerException, InterpreterException { void newObject() { assertNotNull(hbaseInterpreter); } + + @Test + void testDefaultCommandTimeout() { + HbaseInterpreter interpreter = new HbaseInterpreter(new Properties()); + + assertEquals("600000", interpreter.getProperty("shell.command.timeout.millisecs")); + } + + @Test + void testCommandTimeoutIsNotOverridden() { + Properties properties = new Properties(); + properties.put("shell.command.timeout.millisecs", "1000"); + + HbaseInterpreter interpreter = new HbaseInterpreter(properties); + + assertEquals("1000", interpreter.getProperty("shell.command.timeout.millisecs")); + } + + @Test + void testHbaseHomeFallback() { + assertEquals("/usr/lib/hbase", new HbaseInterpreter(new Properties()).getHbaseHome()); + + Properties legacyProperties = new Properties(); + legacyProperties.put("hbase.home", "/opt/hbase"); + assertEquals("/opt/hbase", new HbaseInterpreter(legacyProperties).getHbaseHome()); + + Properties bothProperties = new Properties(); + bothProperties.put("hbase.home", "/opt/hbase"); + bothProperties.put("HBASE_HOME", "/usr/local/hbase"); + assertEquals("/usr/local/hbase", new HbaseInterpreter(bothProperties).getHbaseHome()); + + Properties blankProperties = new Properties(); + blankProperties.put("hbase.home", "/opt/hbase"); + blankProperties.put("HBASE_HOME", " "); + assertEquals("/opt/hbase", new HbaseInterpreter(blankProperties).getHbaseHome()); + } + + @Test + void testShellQuote() { + assertEquals("'list'", HbaseInterpreter.shellQuote("list")); + assertEquals("'a'\\''b'", HbaseInterpreter.shellQuote("a'b")); + assertEquals("'\"quoted\"'", HbaseInterpreter.shellQuote("\"quoted\"")); + assertEquals("'$100'", HbaseInterpreter.shellQuote("$100")); + } + + @Test + void testBuildCommand() { + HbaseInterpreter interpreter = new HbaseInterpreter(new Properties()); + + assertEquals("printf '%s\\n' 'list' | '/usr/lib/hbase/bin/hbase' shell -n", + interpreter.buildCommand("list")); + } + + @Test + void testBuildCommandQuotesSpecialCharacters() { + HbaseInterpreter interpreter = new HbaseInterpreter(new Properties()); + + assertEquals("printf '%s\\n' " + + "'scan '\\''t'\\'', {FILTER => \"PrefixFilter('\\''a'\\'')\"}'" + + " | '/usr/lib/hbase/bin/hbase' shell -n", + interpreter.buildCommand("scan 't', {FILTER => \"PrefixFilter('a')\"}")); + + assertEquals("printf '%s\\n' " + + "'put '\\''t'\\'','\\''r'\\'','\\''c'\\'','\\''$100'\\'''" + + " | '/usr/lib/hbase/bin/hbase' shell -n", + interpreter.buildCommand("put 't','r','c','$100'")); + } + + @Test + void testBuildCommandQuotesHbaseHome() { + Properties properties = new Properties(); + properties.put("HBASE_HOME", "/opt/hbase home"); + HbaseInterpreter interpreter = new HbaseInterpreter(properties); + + assertEquals("printf '%s\\n' 'list' | '/opt/hbase home/bin/hbase' shell -n", + interpreter.buildCommand("list")); + } + + @Test + void testBlankCommandIsNoop() { + InterpreterContext context = new InterpreterContext.Builder() + .setNoteId("note1") + .setParagraphId("paragraph_1") + .build(); + + assertEquals(InterpreterResult.Code.SUCCESS, + hbaseInterpreter.internalInterpret(" ", context).code()); + } } diff --git a/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/HBaseIntegrationTest.java b/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/HBaseIntegrationTest.java index b8e7e18693a..2382eb1f418 100644 --- a/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/HBaseIntegrationTest.java +++ b/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/HBaseIntegrationTest.java @@ -75,7 +75,7 @@ void setup() { @Test public void testHBaseInterpreter() throws Exception { InterpreterSetting hbaseInterpreterSetting = interpreterSettingManager.getInterpreterSettingByName("hbase"); - hbaseInterpreterSetting.setProperty("hbase.home", hbaseHome); + hbaseInterpreterSetting.setProperty("HBASE_HOME", hbaseHome); Interpreter hbaseInterpreter = interpreterFactory.getInterpreter("hbase", new ExecutionContext("user1", "note1", "hbase"));