This is an automated email from the ASF dual-hosted git repository.
hansva pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new f15717e1b9 Issue #6978 (#6979)
f15717e1b9 is described below
commit f15717e1b9b2cd2e7e87f45e571f5988cbc7fb3d
Author: Matt Casters <[email protected]>
AuthorDate: Thu Apr 9 14:57:58 2026 +0200
Issue #6978 (#6979)
* issue #6978 (PyHop server token and stopping)
* issue #6978 (PyHop server token and stopping)
---
.../pages/hop-tools/hop-python/hop-python.adoc | 67 +++++++++++++++++++++-
.../pipeline/beam/getting-started-with-beam.adoc | 5 ++
.../src/main/java/org/apache/hop/py4j/PyHop.java | 31 +++++++++-
.../java/org/apache/hop/py4j/PythonCommand.java | 35 +++++++----
4 files changed, 124 insertions(+), 14 deletions(-)
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/hop-tools/hop-python/hop-python.adoc
b/docs/hop-user-manual/modules/ROOT/pages/hop-tools/hop-python/hop-python.adoc
index 3124907c58..15f10f60fc 100644
---
a/docs/hop-user-manual/modules/ROOT/pages/hop-tools/hop-python/hop-python.adoc
+++
b/docs/hop-user-manual/modules/ROOT/pages/hop-tools/hop-python/hop-python.adoc
@@ -43,6 +43,39 @@ $ sh hop python
2026/04/03 21:03:48 - HopPython - The Hop Python Gateway server was started on
127.0.0.1:25333
----
+=== Usage
+
+[source,bash]
+----
+$ sh hop python --help
+Usage: hop python [-hV] [-e=<environmentOption>]
+ [--gateway-ip-address=<gatewayAddress>]
+ [--gateway-port=<gatewayPort>]
+ [--gateway-stop-password=<stopPassword>]
+ [--gateway-token=<gatewayToken>] [-j=<projectOption>]
+Run the Hop Python gateway (py4j)
+ -e, --environment=<environmentOption>
+ The name of the lifecycle environment to use
+ --gateway-ip-address=<gatewayAddress>
+ The server on which to run the Hop Python (py4j) gateway
+ service. The default is 127.0.0.1 (localhost). Use
+ 0.0.0.0 to make the service widely available.
+ --gateway-port=<gatewayPort>
+ The port on which to run the Hop Python (py4j) gateway
+ service. The default port is 25333.
+ --gateway-stop-password=<stopPassword>
+ If you specify this password it can be used when halting the
+ server in a Python script. Without a password the server
+ can not be stopped this way.
+ --gateway-token=<gatewayToken>
+ Only allow connections to the Hop Python (py4j) gateway that
+ provide this token
+ -h, --help Show this help message and exit.
+ -j, --project=<projectOption>
+ The name of the project to use
+ -V, --version Print version information and exit.
+----
+
=== Ports and addresses
To run the gateway on different ports or make it available on different IP
addresses, you can use these options:
@@ -57,6 +90,12 @@ To run the gateway on different ports or make it available
on different IP addre
|`--gateway-ip-address`
|The IP address on which to run the Hop Python (py4j) gateway service. The
default is `127.0.0.1` (localhost). Use `0.0.0.0` to make the service widely
available.
+|`--gateway-token`
+|Only allow connections to the Hop Python (py4j) gateway that provide this
token.
+
+|`--gateway-stop-password`
+|If you specify this password it can be used when stopping the server
xref:hop-tools/hop-python/hop-python.adoc#_stopping_the_server[in a Python
script]. Without a password the server can't be stopped this way.
+
|===
@@ -127,10 +166,23 @@ You can create gateway parameters if you deviate from the
default port (25333) o
[source,python]
----
+from py4j.java_gateway import JavaGateway, GatewayParameters
gateway_params = GatewayParameters(address='192.168.1.50', port=25333)
gateway = JavaGateway(gateway_parameters=gateway_params)
----
+=== Connecting with a token
+
+If you started the Hop Python server with a token, you need to specify it in
your script:
+
+[source,python]
+----
+from py4j.java_gateway import JavaGateway, GatewayParameters
+params = GatewayParameters(auth_token="YourToken")
+gateway = JavaGateway(gateway_parameters=params)
+----
+
+
== The basics
=== Building a pipeline
@@ -285,7 +337,7 @@ dbSerializer = hop.getMetadataSerializer("rdbms")
dwh = dbSerializer.load("DWH")
----
-If you want to get fancy and work with all the metadata plugin keys, you can
use method `listMetadataKeys()` which gives you an array of strings. Likewise,
you can get a list of database element names as an array with
`listMetadataElements("rdbms")`
+If you want to get fancy and work with all the metadata plugin keys, you can
use method `listMetadataKeys()` which gives you an array of strings.Likewise,
you can get a list of database element names as an array with
`listMetadataElements("rdbms")`
=== Creating new metadata elements
@@ -341,6 +393,19 @@ csv = t.getTransform()
field = hop.newTransformObject(csv,
"org.apache.hop.pipeline.transforms.csvinput.CsvInputField")
----
+[#_stopping_the_server]
+=== Stopping the server
+
+If you're writing tests in Python and don't need the server anymore, you can
stop it using:
+
+[source,python]
+----
+hop.stopServer("your-stop-password")
+----
+
+The stop server password can be specified as an option during startup of the
server.
+
+
== Examples
=== Create a pipeline from scratch and run it
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/beam/getting-started-with-beam.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/beam/getting-started-with-beam.adoc
index 467b85ff41..0b61f1b47f 100644
---
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/beam/getting-started-with-beam.adoc
+++
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/beam/getting-started-with-beam.adoc
@@ -52,6 +52,11 @@ Here is the documentation for the relevant plugins:
|===
|Hop version |Beam version |Spark version |Flink version
+|2.18.0
+|2.71.0
+|3.5.X (scala 2.12)
+|1.19.x
+
|2.17.0
|2.70.0
|3.5.X (scala 2.12)
diff --git a/plugins/misc/py4j/src/main/java/org/apache/hop/py4j/PyHop.java
b/plugins/misc/py4j/src/main/java/org/apache/hop/py4j/PyHop.java
index c13984092a..9b7e1724ae 100644
--- a/plugins/misc/py4j/src/main/java/org/apache/hop/py4j/PyHop.java
+++ b/plugins/misc/py4j/src/main/java/org/apache/hop/py4j/PyHop.java
@@ -20,6 +20,7 @@ package org.apache.hop.py4j;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.ArrayBlockingQueue;
import lombok.Getter;
import lombok.Setter;
import org.apache.commons.lang3.StringUtils;
@@ -35,6 +36,7 @@ import org.apache.hop.core.plugins.IPlugin;
import org.apache.hop.core.plugins.IPluginType;
import org.apache.hop.core.plugins.PluginRegistry;
import org.apache.hop.core.plugins.TransformPluginType;
+import org.apache.hop.core.util.StringUtil;
import org.apache.hop.core.variables.IVariables;
import org.apache.hop.metadata.api.HopMetadata;
import org.apache.hop.metadata.api.IHopMetadata;
@@ -62,8 +64,12 @@ public class PyHop {
private IVariables variables;
private IHopMetadataProvider metadataProvider;
private ILogChannel log;
+ private String stopPassword;
+ private ArrayBlockingQueue<Object> blockingQueue;
- public PyHop() {}
+ public PyHop() {
+ this.blockingQueue = new ArrayBlockingQueue<>(10);
+ }
public void initialize(
IVariables variables, IHopMetadataProvider metadataProvider, ILogChannel
log) {
@@ -432,12 +438,33 @@ public class PyHop {
return loggingBuffer.getBuffer(logChannelId, false).toString();
}
+ //
+ // The PyHop server
+ //
+
+ public void stopServer(String passwordToVerify) throws HopException {
+ // Only with a stop password can the server be stopped.
+ if (!StringUtil.isEmpty(this.stopPassword) &&
this.stopPassword.equals(passwordToVerify)) {
+ blockingQueue.add(new Object());
+ } else {
+ throw new HopException("Specify a stop password, here and during server
startup.");
+ }
+ }
+
+ public void waitUntilStopped() {
+ try {
+ blockingQueue.take();
+ } catch (InterruptedException e) {
+ log.logError("Waiting interrupted on the PyHop gateway server", e);
+ }
+ }
+
//
// Utility
//
public static @NonNull String describeAvailablePlugins(
- Class<? extends IPluginType> pluginTypeClass) {
+ Class<? extends IPluginType<?>> pluginTypeClass) {
List<IPlugin> plugins =
PluginRegistry.getInstance().getPlugins(pluginTypeClass);
StringBuilder available = new StringBuilder();
for (IPlugin plugin : plugins) {
diff --git
a/plugins/misc/py4j/src/main/java/org/apache/hop/py4j/PythonCommand.java
b/plugins/misc/py4j/src/main/java/org/apache/hop/py4j/PythonCommand.java
index 0f7042908c..1f29d97679 100644
--- a/plugins/misc/py4j/src/main/java/org/apache/hop/py4j/PythonCommand.java
+++ b/plugins/misc/py4j/src/main/java/org/apache/hop/py4j/PythonCommand.java
@@ -18,8 +18,6 @@
package org.apache.hop.py4j;
-import static java.lang.Thread.sleep;
-
import java.net.InetAddress;
import java.util.Map;
import lombok.Getter;
@@ -28,6 +26,7 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.hop.core.Const;
import org.apache.hop.core.config.plugin.ConfigPlugin;
import org.apache.hop.core.config.plugin.IConfigOptions;
+import org.apache.hop.core.encryption.Encr;
import org.apache.hop.core.exception.HopException;
import org.apache.hop.core.logging.ILogChannel;
import org.apache.hop.core.logging.LogChannel;
@@ -65,9 +64,23 @@ public class PythonCommand implements Runnable, IHopCommand,
IHasHopMetadataProv
"The server on which to run the Hop Python (py4j) gateway service.
The default is 127.0.0.1 (localhost). Use 0.0.0.0 to make the service widely
available.")
private String gatewayAddress;
+ @CommandLine.Option(
+ names = {"--gateway-token"},
+ description =
+ "Only allow connections to the Hop Python (py4j) gateway that
provide this token")
+ private String gatewayToken;
+
+ @CommandLine.Option(
+ names = {"--gateway-stop-password"},
+ description =
+ "If you specify this password it can be used when halting the server
in a Python script. Without a password the server can not be stopped this
way.")
+ private String stopPassword;
+
private PyHop pyHop;
- public PythonCommand() {}
+ public PythonCommand() {
+ // Empty by design
+ }
@Override
public void initialize(
@@ -130,21 +143,21 @@ public class PythonCommand implements Runnable,
IHopCommand, IHasHopMetadataProv
if (StringUtils.isEmpty(ipAddress)) {
ipAddress = "127.0.0.1";
}
+ String token =
Encr.decryptPasswordOptionallyEncrypted(variables.resolve(gatewayToken));
// Run the gateway
//
GatewayServer.GatewayServerBuilder builder = new
GatewayServer.GatewayServerBuilder();
- GatewayServer gatewayServer =
- builder
- .entryPoint(this)
- .javaPort(port)
- .javaAddress(InetAddress.getByName(ipAddress))
- .build();
+ builder =
+
builder.entryPoint(this).javaPort(port).javaAddress(InetAddress.getByName(ipAddress));
+ if (StringUtils.isNotEmpty(token)) {
+ builder.authToken(token);
+ }
+ GatewayServer gatewayServer = builder.build();
gatewayServer.start();
log.logBasic("The Hop Python Gateway server was started on " + ipAddress
+ ":" + port);
- do sleep(100);
- while (true);
+ pyHop.waitUntilStopped();
} catch (Exception e) {
log.logError("Error running the Hop Python Gateway server (py4j)", e);
}
