Skip to content

[ZEPPELIN-6770] Update HbaseInterpreter to a subclass of ShellInterpr… #5613

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
paul8263 wants to merge 1 commit into
base: master
Choose a base branch
from
+157 −162
Open

[ZEPPELIN-6770] Update HbaseInterpreter to a subclass of ShellInterpr… #5613

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion docs/interpreter/hbase.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,10 +39,15 @@ Zeppelin is built against HBase 1.x and 2.x releases.
<th>Description</th>
</tr>
<tr>
<td>hbase.home</td>
<td>HBASE_HOME</td>
<td>/usr/lib/hbase</td>
<td>Installation directory of HBase, defaults to HBASE_HOME in environment</td>
</tr>
<tr>
<td>shell.command.timeout.millisecs</td>
<td>600000</td>
<td>HBase shell command timeout in millisecs when there is no output, defaults to 10 minutes</td>
</tr>
</table>

If you want to connect to HBase running on a cluster, you'll need to follow the next step.
Expand Down
9 changes: 3 additions & 6 deletions hbase/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,12 +37,9 @@

<dependencies>
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<groupId>org.apache.zeppelin</groupId>
<artifactId>zeppelin-shell</artifactId>

@jongyoul jongyoul Nov 12, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@paul8263 I know it's the proper way to extend the existing feature but I wouldn't recommend to include another interpreter. It sometimes leads unexpected behaviors as we designed every interpreter as independent component. Could you please mimic and/or re-implement features you use only in this interpreter? I'm also planning to existing dependencies between interpreters in the following year.

<version>${project.version}</version>
</dependency>
</dependencies>

Expand Down
189 changes: 44 additions & 145 deletions hbase/src/main/java/org/apache/zeppelin/hbase/HbaseInterpreter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<String, Executor> runningProcesses = new HashMap<>();
import org.apache.zeppelin.shell.ShellInterpreter;

private Map<String, File> 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();
}
}
12 changes: 9 additions & 3 deletions hbase/src/main/resources/interpreter-setting.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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": {
Expand Down
100 changes: 94 additions & 6 deletions hbase/src/test/java/org/apache/zeppelin/hbase/HbaseInterpreterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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();
Expand All @@ -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());
}
}
Loading
Loading