You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hop.apache.org by ha...@apache.org on 2022/04/08 06:47:46 UTC
[hop] branch master updated: HOP-3865: make DynamicWaitTime configurable and lower the default value
This is an automated email from the ASF dual-hosted git repository.
hansva pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/master by this push:
new 0d1223c2cf HOP-3865: make DynamicWaitTime configurable and lower the default value
new d2fd53875f Merge pull request #1447 from hansva/HOP-3865
0d1223c2cf is described below
commit 0d1223c2cff483f3c2f5d71d89fc9ebace6674a7
Author: Hans Van Akelyen <ha...@gmail.com>
AuthorDate: Thu Apr 7 13:02:22 2022 +0200
HOP-3865: make DynamicWaitTime configurable and lower the default value
HOP-3865
---
core/src/main/java/org/apache/hop/core/Const.java | 295 +++++++++++++++------
.../native-local-pipeline-engine.adoc | 4 +
.../local/LocalPipelineRunConfiguration.java | 72 ++++-
.../hop/pipeline/transform/BaseTransform.java | 177 +++++++++----
.../hop/pipeline/transform/DynamicWaitTimes.java | 23 +-
.../pipeline/transform/DynamicWaitTimesTest.java | 6 +-
.../metadata/pipeline-run-configuration/local.json | 4 +-
.../pipeline/transforms/xml/addxml/AddXmlTest.java | 2 +
.../config/messages/messages_en_US.properties | 1 +
9 files changed, 437 insertions(+), 147 deletions(-)
diff --git a/core/src/main/java/org/apache/hop/core/Const.java b/core/src/main/java/org/apache/hop/core/Const.java
index cc92727d2a..677711ed8f 100644
--- a/core/src/main/java/org/apache/hop/core/Const.java
+++ b/core/src/main/java/org/apache/hop/core/Const.java
@@ -105,7 +105,10 @@ public class Const {
/*
The name of the history folder in which the Hop local audit manager saves data
*/
- @Variable(scope=VariableScope.SYSTEM, description="The name of the history folder in which the Hop local audit manager saves data.")
+ @Variable(
+ scope = VariableScope.SYSTEM,
+ description =
+ "The name of the history folder in which the Hop local audit manager saves data.")
public static final String HOP_AUDIT_FOLDER =
NVL(
System.getProperty("HOP_AUDIT_FOLDER"),
@@ -114,7 +117,10 @@ public class Const {
/*
The name of the history folder in which the Hop local audit manager saves data
*/
- @Variable(scope=VariableScope.SYSTEM, description="The name of the history folder in which the Hop local audit manager saves data.")
+ @Variable(
+ scope = VariableScope.SYSTEM,
+ description =
+ "The name of the history folder in which the Hop local audit manager saves data.")
public static final String HOP_CONFIG_FOLDER =
NVL(
System.getProperty("HOP_CONFIG_FOLDER"),
@@ -124,28 +130,38 @@ public class Const {
* The name of the variable which configures whether or not we should automatically create a
* config file when it's missing
*/
- @Variable(scope=VariableScope.SYSTEM, value="N", description="Set this variable to 'Y' to automatically create config file when it's missing.")
+ @Variable(
+ scope = VariableScope.SYSTEM,
+ value = "N",
+ description =
+ "Set this variable to 'Y' to automatically create config file when it's missing.")
public static final String HOP_AUTO_CREATE_CONFIG = "HOP_AUTO_CREATE_CONFIG";
/**
* The system environment variable pointing to the alternative location for the Hop metadata
* folder
*/
- @Variable(scope=VariableScope.SYSTEM, description="The variable which points to the alternative location for the Hop metadata.")
+ @Variable(
+ scope = VariableScope.SYSTEM,
+ description = "The variable which points to the alternative location for the Hop metadata.")
public static final String HOP_METADATA_FOLDER = "HOP_METADATA_FOLDER";
/** The variable which points to a shared folder with JDBC drivers in them. */
- @Variable(scope=VariableScope.SYSTEM, description="The variable which points to a shared folder with JDBC drivers in them.")
+ @Variable(
+ scope = VariableScope.SYSTEM,
+ description = "The variable which points to a shared folder with JDBC drivers in them.")
public static final String HOP_SHARED_JDBC_FOLDER = "HOP_SHARED_JDBC_FOLDER";
/** The operating system the hop platform runs on */
- @Variable(scope=VariableScope.SYSTEM, description="The operating system the hop platform runs on.")
+ @Variable(
+ scope = VariableScope.SYSTEM,
+ description = "The operating system the hop platform runs on.")
public static final String HOP_PLATFORM_OS = "HOP_PLATFORM_OS";
/** The runtime that is being used */
- @Variable(scope=VariableScope.SYSTEM, description="The runtime that is being used.")
- public static final String HOP_PLATFORM_RUNTIME = "HOP_PLATFORM_RUNTIME";
-
+ @Variable(scope = VariableScope.SYSTEM, description = "The runtime that is being used.")
+ public static final String HOP_PLATFORM_RUNTIME = "HOP_PLATFORM_RUNTIME";
+
/** An empty ("") String. */
public static final String EMPTY_STRING = "";
@@ -288,7 +304,7 @@ public class Const {
Const.INTERNAL_VARIABLE_WORKFLOW_FILENAME_NAME,
Const.INTERNAL_VARIABLE_WORKFLOW_NAME,
};
-
+
/** The pipeline filename directory */
public static final String INTERNAL_VARIABLE_PIPELINE_FILENAME_DIRECTORY =
INTERNAL_VARIABLE_PREFIX + ".Pipeline.Filename.Directory";
@@ -347,24 +363,33 @@ public class Const {
/** The default undo level for Hop */
public static final int MAX_UNDO = 100;
-
+
/**
* If you set this environment variable you can limit the log size of all pipelines and workflows
* that don't have the "log size limit" property set in their respective properties.
*/
- @Variable(value="0", description="The log size limit for all pipelines and workflows that don't have the \"log size limit\" property set in their respective properties.")
+ @Variable(
+ value = "0",
+ description =
+ "The log size limit for all pipelines and workflows that don't have the \"log size limit\" property set in their respective properties.")
public static final String HOP_LOG_SIZE_LIMIT = "HOP_LOG_SIZE_LIMIT";
/** The name of the variable that defines the timer used for detecting server nodes. */
- @Variable(description="The name of the variable that defines the timer used for detecting server nodes")
+ @Variable(
+ description =
+ "The name of the variable that defines the timer used for detecting server nodes")
public static final String HOP_SERVER_DETECTION_TIMER = "HOP_SERVER_DETECTION_TIMER";
/**
* System wide flag to drive the evaluation of null in ValueMeta. If this setting is set to "Y",
* an empty string and null are different. Otherwise they are not.
*/
- @Variable(value="N", description="NULL vs Empty String. If this setting is set to 'Y', an empty string and null are different. Otherwise they are not")
- public static final String HOP_EMPTY_STRING_DIFFERS_FROM_NULL = "HOP_EMPTY_STRING_DIFFERS_FROM_NULL";
+ @Variable(
+ value = "N",
+ description =
+ "NULL vs Empty String. If this setting is set to 'Y', an empty string and null are different. Otherwise they are not")
+ public static final String HOP_EMPTY_STRING_DIFFERS_FROM_NULL =
+ "HOP_EMPTY_STRING_DIFFERS_FROM_NULL";
/**
* System wide flag to allow non-strict string to number conversion for backward compatibility. If
@@ -373,7 +398,10 @@ public class Const {
* symbol). The default (N) will be to throw an error if non-numeric symbols are found in the
* string.
*/
- @Variable(value="N", description="System wide flag to allow lenient string to number conversion for backward compatibility. If this setting is set to 'Y', an string starting with digits will be converted successfully into a number. (example: 192.168.1.1 will be converted into 192 or 192.168 or 192168 depending on the decimal and grouping symbol). The default (N) will be to throw an error if non-numeric symbols are found in the string.")
+ @Variable(
+ value = "N",
+ description =
+ "System wide flag to allow lenient string to number conversion for backward compatibility. If this setting is set to 'Y', an string starting with digits will be converted successfully into a number. (example: 192.168.1.1 will be converted into 192 or 192.168 or 192168 depending on the decimal and grouping symbol). The default (N) will be to throw an error if non-numeric symbols are found in the string.")
public static final String HOP_LENIENT_STRING_TO_NUMBER_CONVERSION =
"HOP_LENIENT_STRING_TO_NUMBER_CONVERSION";
@@ -381,39 +409,59 @@ public class Const {
* You can use this variable to speed up hostname lookup. Hostname lookup is performed by Hop so
* that it is capable of logging the server on which a workflow or pipeline is executed.
*/
- @Variable(description="You can use this variable to speed up hostname lookup. Hostname lookup is performed by Hop so that it is capable of logging the server on which a workflow or pipeline is executed.")
+ @Variable(
+ description =
+ "You can use this variable to speed up hostname lookup. Hostname lookup is performed by Hop so that it is capable of logging the server on which a workflow or pipeline is executed.")
public static final String HOP_SYSTEM_HOSTNAME = "HOP_SYSTEM_HOSTNAME";
/**
* System wide flag to set the maximum number of log lines that are kept internally by Hop. Set to
* 0 to keep all rows (default)
*/
- @Variable(scope=VariableScope.APPLICATION, value="0", description="The maximum number of log lines that are kept internally by Hop. Set to 0 to keep all rows (default)")
+ @Variable(
+ scope = VariableScope.APPLICATION,
+ value = "0",
+ description =
+ "The maximum number of log lines that are kept internally by Hop. Set to 0 to keep all rows (default)")
public static final String HOP_MAX_LOG_SIZE_IN_LINES = "HOP_MAX_LOG_SIZE_IN_LINES";
/**
* System wide flag to set the maximum age (in minutes) of a log line while being kept internally
* by Hop. Set to 0 to keep all rows indefinitely (default)
*/
- @Variable(scope=VariableScope.APPLICATION, value="1440", description="The maximum age (in minutes) of a log line while being kept internally by Hop. Set to 0 to keep all rows indefinitely (default)")
+ @Variable(
+ scope = VariableScope.APPLICATION,
+ value = "1440",
+ description =
+ "The maximum age (in minutes) of a log line while being kept internally by Hop. Set to 0 to keep all rows indefinitely (default)")
public static final String HOP_MAX_LOG_TIMEOUT_IN_MINUTES = "HOP_MAX_LOG_TIMEOUT_IN_MINUTES";
/**
* System wide flag to determine whether standard error will be redirected to Hop logging
* facilities. Will redirect if the value is equal ignoring case to the string "Y"
*/
- @Variable(scope=VariableScope.SYSTEM, value="N", description="Set this variable to 'Y' to redirect stderr to Hop logging.")
+ @Variable(
+ scope = VariableScope.SYSTEM,
+ value = "N",
+ description = "Set this variable to 'Y' to redirect stderr to Hop logging.")
public static final String HOP_REDIRECT_STDERR = "HOP_REDIRECT_STDERR";
/**
* System wide flag to determine whether standard out will be redirected to Hop logging
* facilities. Will redirect if the value is equal ignoring case to the string "Y"
*/
- @Variable(scope=VariableScope.SYSTEM, value="N", description="Set this variable to 'Y' to redirect stdout to Hop logging.")
+ @Variable(
+ scope = VariableScope.SYSTEM,
+ value = "N",
+ description = "Set this variable to 'Y' to redirect stdout to Hop logging.")
public static final String HOP_REDIRECT_STDOUT = "HOP_REDIRECT_STDOUT";
/** System wide flag to log stack traces in a simpler, more human readable format */
- @Variable(scope=VariableScope.SYSTEM, value="N", description="Set this variable to 'Y' to log stack traces in a simpler, more human readable format.")
+ @Variable(
+ scope = VariableScope.SYSTEM,
+ value = "N",
+ description =
+ "Set this variable to 'Y' to log stack traces in a simpler, more human readable format.")
public static final String HOP_SIMPLE_STACK_TRACES = "HOP_SIMPLE_STACK_TRACES";
public static final boolean isUsingSimpleStackTraces() {
@@ -425,7 +473,10 @@ public class Const {
* This environment variable will set a time-out after which waiting, completed or stopped
* pipelines and workflows will be automatically cleaned up. The default value is 1440 (one day).
*/
- @Variable(value="1440", description="This project variable will set a time-out after which waiting, completed or stopped pipelines and workflows will be automatically cleaned up. The default value is 1440 (one day).")
+ @Variable(
+ value = "1440",
+ description =
+ "This project variable will set a time-out after which waiting, completed or stopped pipelines and workflows will be automatically cleaned up. The default value is 1440 (one day).")
public static final String HOP_SERVER_OBJECT_TIMEOUT_MINUTES =
"HOP_SERVER_OBJECT_TIMEOUT_MINUTES";
@@ -433,47 +484,66 @@ public class Const {
* System wide parameter: the maximum number of transform performance snapshots to keep in memory.
* Set to 0 to keep all snapshots indefinitely (default)
*/
- @Variable(value="0", description="The maximum number of transform performance snapshots to keep in memory. Set to 0 to keep all snapshots indefinitely (default)")
+ @Variable(
+ value = "0",
+ description =
+ "The maximum number of transform performance snapshots to keep in memory. Set to 0 to keep all snapshots indefinitely (default)")
public static final String HOP_TRANSFORM_PERFORMANCE_SNAPSHOT_LIMIT =
"HOP_TRANSFORM_PERFORMANCE_SNAPSHOT_LIMIT";
/** A variable to configure the maximum number of workflow trackers kept in memory. */
- @Variable(value="5000", description="The maximum age (in minutes) of a log line while being kept internally by Hop. Set to 0 to keep all rows indefinitely (default)")
+ @Variable(
+ value = "5000",
+ description =
+ "The maximum age (in minutes) of a log line while being kept internally by Hop. Set to 0 to keep all rows indefinitely (default)")
public static final String HOP_MAX_WORKFLOW_TRACKER_SIZE = "HOP_MAX_WORKFLOW_TRACKER_SIZE";
/**
* A variable to configure the maximum number of action results kept in memory for logging
* purposes.
*/
- @Variable(value="5000", description="The maximum number of action results kept in memory for logging purposes.")
+ @Variable(
+ value = "5000",
+ description = "The maximum number of action results kept in memory for logging purposes.")
public static final String HOP_MAX_ACTIONS_LOGGED = "HOP_MAX_ACTIONS_LOGGED";
/**
* A variable to configure the maximum number of logging registry entries kept in memory for
* logging purposes.
*/
- @Variable(value="10000", description="The maximum number of logging registry entries kept in memory for logging purposes")
+ @Variable(
+ value = "10000",
+ description =
+ "The maximum number of logging registry entries kept in memory for logging purposes")
public static final String HOP_MAX_LOGGING_REGISTRY_SIZE = "HOP_MAX_LOGGING_REGISTRY_SIZE";
/** A variable to configure the hop log tab refresh delay. */
- @Variable(scope=VariableScope.APPLICATION, value="1000", description="The hop log tab refresh delay.")
+ @Variable(
+ scope = VariableScope.APPLICATION,
+ value = "1000",
+ description = "The hop log tab refresh delay.")
public static final String HOP_LOG_TAB_REFRESH_DELAY = "HOP_LOG_TAB_REFRESH_DELAY";
/** A variable to configure the hop log tab refresh period. */
- @Variable(scope=VariableScope.APPLICATION, value="1000", description="The hop log tab refresh period.")
+ @Variable(
+ scope = VariableScope.APPLICATION,
+ value = "1000",
+ description = "The hop log tab refresh period.")
public static final String HOP_LOG_TAB_REFRESH_PERIOD = "HOP_LOG_TAB_REFRESH_PERIOD";
/**
* Name of the environment variable to specify additional classes to scan for plugin annotations
*/
- @Variable(description="A comma delimited list of classes to scan for plugin annotations")
+ @Variable(description = "A comma delimited list of classes to scan for plugin annotations")
public static final String HOP_PLUGIN_CLASSES = "HOP_PLUGIN_CLASSES";
/**
* Name of the environment variable to specify additional packaged to scan for plugin annotations
* (warning: slow!)
*/
- @Variable(description="A comma delimited list of packages to scan for plugin annotations (warning: slow!!)")
+ @Variable(
+ description =
+ "A comma delimited list of packages to scan for plugin annotations (warning: slow!!)")
public static final String HOP_PLUGIN_PACKAGES = "HOP_PLUGIN_PACKAGES";
public static final String HOP_PLUGIN_BASE_FOLDERS = "HOP_PLUGIN_BASE_FOLDERS";
@@ -482,7 +552,9 @@ public class Const {
* Name of the environment variable that contains the size of the pipeline rowset size. This
* overwrites values that you set pipeline settings.
*/
- @Variable(description="Name of the environment variable that contains the size of the pipeline rowset size. This overwrites values that you set pipeline settings")
+ @Variable(
+ description =
+ "Name of the environment variable that contains the size of the pipeline rowset size. This overwrites values that you set pipeline settings")
public static final String HOP_PIPELINE_ROWSET_SIZE = "HOP_PIPELINE_ROWSET_SIZE";
/** A general initial version comment */
@@ -492,7 +564,9 @@ public class Const {
public static final String VERSION_COMMENT_EDIT_VERSION = "Modification by user";
/** Specifies the password encoding plugin to use by ID (Hop is the default). */
- @Variable(value="Hop", description="Specifies the password encoder plugin to use by ID (Hop is the default).")
+ @Variable(
+ value = "Hop",
+ description = "Specifies the password encoder plugin to use by ID (Hop is the default).")
public static final String HOP_PASSWORD_ENCODER_PLUGIN = "HOP_PASSWORD_ENCODER_PLUGIN";
/**
@@ -505,76 +579,108 @@ public class Const {
* The name of the variable that optionally contains an alternative rowset get timeout (in ms).
* This only makes a difference for extremely short lived pipelines.
*/
- @Variable(value = "50",
- description = "The name of the variable that optionally contains an alternative rowset get timeout (in ms). This only makes a difference for extremely short lived pipelines.")
+ @Variable(
+ value = "50",
+ description =
+ "The name of the variable that optionally contains an alternative rowset get timeout (in ms). This only makes a difference for extremely short lived pipelines.")
public static final String HOP_ROWSET_GET_TIMEOUT = "HOP_ROWSET_GET_TIMEOUT";
/**
* The name of the variable that optionally contains an alternative rowset put timeout (in ms).
* This only makes a difference for extremely short lived pipelines.
*/
- @Variable(value = "50",
- description = "The name of the variable that optionally contains an alternative rowset put timeout (in ms). This only makes a difference for extremely short lived pipelines.")
+ @Variable(
+ value = "50",
+ description =
+ "The name of the variable that optionally contains an alternative rowset put timeout (in ms). This only makes a difference for extremely short lived pipelines.")
public static final String HOP_ROWSET_PUT_TIMEOUT = "HOP_ROWSET_PUT_TIMEOUT";
/** Set this variable to Y if you want to test a more efficient batching row set. (default = N) */
- @Variable(value="N", description="Set this variable to 'Y' if you want to test a more efficient batching row set.")
+ @Variable(
+ value = "N",
+ description =
+ "Set this variable to 'Y' if you want to test a more efficient batching row set.")
public static final String HOP_BATCHING_ROWSET = "HOP_BATCHING_ROWSET";
/**
* Set this variable to limit max number of files the Text File Output transform can have open at
* one time.
*/
- @Variable(value="1024", description="This project variable is used by the Text File Output transform. It defines the max number of simultaneously open files within the transform. The transform will close/reopen files as necessary to insure the max is not exceeded")
+ @Variable(
+ value = "1024",
+ description =
+ "This project variable is used by the Text File Output transform. It defines the max number of simultaneously open files within the transform. The transform will close/reopen files as necessary to insure the max is not exceeded")
public static final String HOP_FILE_OUTPUT_MAX_STREAM_COUNT = "HOP_FILE_OUTPUT_MAX_STREAM_COUNT";
/**
* This variable contains the number of milliseconds between flushes of all open files in the Text
* File Output transform.
*/
- @Variable(value="0", description="This project variable is used by the Text File Output transform. It defines the max number of milliseconds between flushes of files opened by the transform.")
+ @Variable(
+ value = "0",
+ description =
+ "This project variable is used by the Text File Output transform. It defines the max number of milliseconds between flushes of files opened by the transform.")
public static final String HOP_FILE_OUTPUT_MAX_STREAM_LIFE = "HOP_FILE_OUTPUT_MAX_STREAM_LIFE";
/** Set this variable to Y to disable standard Hop logging to the console. (stdout) */
- @Variable(value="N", description="Set this variable to 'Y' to disable standard Hop logging to the console. (stdout)")
+ @Variable(
+ value = "N",
+ description =
+ "Set this variable to 'Y' to disable standard Hop logging to the console. (stdout)")
public static final String HOP_DISABLE_CONSOLE_LOGGING = "HOP_DISABLE_CONSOLE_LOGGING";
/** The name of the variable containing an alternative default number format */
- @Variable(description="The name of the variable containing an alternative default number format")
+ @Variable(
+ description = "The name of the variable containing an alternative default number format")
public static final String HOP_DEFAULT_NUMBER_FORMAT = "HOP_DEFAULT_NUMBER_FORMAT";
/** The name of the variable containing an alternative default bignumber format */
- @Variable(description="The name of the variable containing an alternative default bignumber format")
+ @Variable(
+ description = "The name of the variable containing an alternative default bignumber format")
public static final String HOP_DEFAULT_BIGNUMBER_FORMAT = "HOP_DEFAULT_BIGNUMBER_FORMAT";
/** The name of the variable containing an alternative default integer format */
- @Variable(description="The name of the variable containing an alternative default integer format")
+ @Variable(
+ description = "The name of the variable containing an alternative default integer format")
public static final String HOP_DEFAULT_INTEGER_FORMAT = "HOP_DEFAULT_INTEGER_FORMAT";
/** The name of the variable containing an alternative default date format */
- @Variable(description="The name of the variable containing an alternative default date format")
+ @Variable(description = "The name of the variable containing an alternative default date format")
public static final String HOP_DEFAULT_DATE_FORMAT = "HOP_DEFAULT_DATE_FORMAT";
// Null values tweaks
- @Variable(value="N", description="Set this variable to 'Y' to set the minimum to NULL if NULL is within an aggregate. Otherwise by default NULL is ignored by the MIN aggregate and MIN is set to the minimum value that is not NULL. See also the variable HOP_AGGREGATION_ALL_NULLS_ARE_ZERO.")
+ @Variable(
+ value = "N",
+ description =
+ "Set this variable to 'Y' to set the minimum to NULL if NULL is within an aggregate. Otherwise by default NULL is ignored by the MIN aggregate and MIN is set to the minimum value that is not NULL. See also the variable HOP_AGGREGATION_ALL_NULLS_ARE_ZERO.")
public static final String HOP_AGGREGATION_MIN_NULL_IS_VALUED =
"HOP_AGGREGATION_MIN_NULL_IS_VALUED";
-
- @Variable(value="N", description="Set this variable to 'Y' to return 0 when all values within an aggregate are NULL. Otherwise by default a NULL is returned when all values are NULL.")
+
+ @Variable(
+ value = "N",
+ description =
+ "Set this variable to 'Y' to return 0 when all values within an aggregate are NULL. Otherwise by default a NULL is returned when all values are NULL.")
public static final String HOP_AGGREGATION_ALL_NULLS_ARE_ZERO =
"HOP_AGGREGATION_ALL_NULLS_ARE_ZERO";
/** The name of the variable containing an alternative default timestamp format */
- @Variable(description="The name of the variable containing an alternative default timestamp format")
+ @Variable(
+ description = "The name of the variable containing an alternative default timestamp format")
public static final String HOP_DEFAULT_TIMESTAMP_FORMAT = "HOP_DEFAULT_TIMESTAMP_FORMAT";
/** Variable that is responsible for removing enclosure symbol after splitting the string */
- @Variable(value="N", description="Set this variable to 'N' to preserve enclosure symbol after splitting the string in the Split fields transform. Changing it to true will remove first and last enclosure symbol from the resulting string chunks.")
+ @Variable(
+ value = "N",
+ description =
+ "Set this variable to 'N' to preserve enclosure symbol after splitting the string in the Split fields transform. Changing it to true will remove first and last enclosure symbol from the resulting string chunks.")
public static final String HOP_SPLIT_FIELDS_REMOVE_ENCLOSURE =
"HOP_SPLIT_FIELDS_REMOVE_ENCLOSURE";
/** Variable that is responsible for checking empty field names and types. */
- @Variable(value="N", description="Set this variable to 'Y' to allow your pipeline to pass 'null' fields and/or empty types.")
+ @Variable(
+ value = "N",
+ description =
+ "Set this variable to 'Y' to allow your pipeline to pass 'null' fields and/or empty types.")
public static final String HOP_ALLOW_EMPTY_FIELD_NAMES_AND_TYPES =
"HOP_ALLOW_EMPTY_FIELD_NAMES_AND_TYPES";
@@ -583,7 +689,10 @@ public class Const {
* Properties -> Log panel. Changing it to true will clear all global log variables when export
* pipeline / workflow
*/
- @Variable(value="N", description="Set this variable to 'N' to preserve global log variables defined in pipeline / workflow Properties -> Log panel. Changing it to 'Y' will clear it when export pipeline / workflow.")
+ @Variable(
+ value = "N",
+ description =
+ "Set this variable to 'N' to preserve global log variables defined in pipeline / workflow Properties -> Log panel. Changing it to 'Y' will clear it when export pipeline / workflow.")
public static final String HOP_GLOBAL_LOG_VARIABLES_CLEAR_ON_EXPORT =
"HOP_GLOBAL_LOG_VARIABLES_CLEAR_ON_EXPORT";
@@ -593,28 +702,35 @@ public class Const {
public static final ReleaseType RELEASE = ReleaseType.GA;
/** A variable to configure turning on/off detailed subjects in log. */
- @Variable(value="N", description="Set this variable to 'Y' to precede transform/action name in log lines with the complete path to the transform/action. Useful to perfectly identify where a problem happened in our process.")
+ @Variable(
+ value = "N",
+ description =
+ "Set this variable to 'Y' to precede transform/action name in log lines with the complete path to the transform/action. Useful to perfectly identify where a problem happened in our process.")
public static final String HOP_LOG_MARK_MAPPINGS = "HOP_LOG_MARK_MAPPINGS";
/** A variable to configure jetty option: acceptors for Hop server */
- @Variable(description="A variable to configure jetty option: acceptors for Hop server")
+ @Variable(description = "A variable to configure jetty option: acceptors for Hop server")
public static final String HOP_SERVER_JETTY_ACCEPTORS = "HOP_SERVER_JETTY_ACCEPTORS";
/** A variable to configure jetty option: acceptQueueSize for Hop server */
- @Variable(description="A variable to configure jetty option: acceptQueueSize for Hop server")
+ @Variable(description = "A variable to configure jetty option: acceptQueueSize for Hop server")
public static final String HOP_SERVER_JETTY_ACCEPT_QUEUE_SIZE =
"HOP_SERVER_JETTY_ACCEPT_QUEUE_SIZE";
/** A variable to configure jetty option: lowResourcesMaxIdleTime for Hop server */
- @Variable(description="A variable to configure jetty option: lowResourcesMaxIdleTime for Hop server")
+ @Variable(
+ description = "A variable to configure jetty option: lowResourcesMaxIdleTime for Hop server")
public static final String HOP_SERVER_JETTY_RES_MAX_IDLE_TIME =
"HOP_SERVER_JETTY_RES_MAX_IDLE_TIME";
- @Variable(description="Defines the default encoding for servlets, leave it empty to use Java default encoding")
+ @Variable(
+ description =
+ "Defines the default encoding for servlets, leave it empty to use Java default encoding")
public static final String HOP_DEFAULT_SERVLET_ENCODING = "HOP_DEFAULT_SERVLET_ENCODING";
-
+
/** A variable to configure refresh for Hop server workflow/pipeline status page */
- @Variable(description="A variable to configure refresh for Hop server workflow/pipeline status page")
+ @Variable(
+ description = "A variable to configure refresh for Hop server workflow/pipeline status page")
public static final String HOP_SERVER_REFRESH_STATUS = "HOP_SERVER_REFRESH_STATUS";
/** A variable to configure s3vfs to use a temporary file on upload data to S3 Amazon." */
@@ -622,7 +738,7 @@ public class Const {
"s3.vfs.useTempFileOnUploadData";
/** A variable to configure Tab size" */
- @Variable(description="A variable to configure Tab size")
+ @Variable(description = "A variable to configure Tab size")
public static final String HOP_MAX_TAB_LENGTH = "HOP_MAX_TAB_LENGTH";
/**
@@ -641,7 +757,9 @@ public class Const {
* @see #HOP_ZIP_MIN_INFLATE_RATIO_DEFAULT
* @see #HOP_ZIP_MIN_INFLATE_RATIO_DEFAULT_STRING
*/
- @Variable(description="A variable to configure the minimum allowed ratio between de- and inflated bytes to detect a zipbomb")
+ @Variable(
+ description =
+ "A variable to configure the minimum allowed ratio between de- and inflated bytes to detect a zipbomb")
public static final String HOP_ZIP_MIN_INFLATE_RATIO = "HOP_ZIP_MIN_INFLATE_RATIO";
/**
@@ -649,7 +767,7 @@ public class Const {
*
* @see #HOP_ZIP_MIN_INFLATE_RATIO
* @see #HOP_ZIP_MIN_INFLATE_RATIO_DEFAULT_STRING
- */
+ */
public static final Double HOP_ZIP_MIN_INFLATE_RATIO_DEFAULT = 0.01d;
/**
@@ -658,7 +776,7 @@ public class Const {
* @see #HOP_ZIP_MIN_INFLATE_RATIO
* @see #HOP_ZIP_MIN_INFLATE_RATIO_DEFAULT
*/
- @Variable(description="")
+ @Variable(description = "")
public static final String HOP_ZIP_MIN_INFLATE_RATIO_DEFAULT_STRING =
String.valueOf(HOP_ZIP_MIN_INFLATE_RATIO_DEFAULT);
@@ -671,7 +789,7 @@ public class Const {
* @see #HOP_ZIP_MAX_ENTRY_SIZE_DEFAULT
* @see #HOP_ZIP_MAX_ENTRY_SIZE_DEFAULT_STRING
*/
- @Variable(description="A variable to configure the maximum file size of a single zip entry")
+ @Variable(description = "A variable to configure the maximum file size of a single zip entry")
public static final String HOP_ZIP_MAX_ENTRY_SIZE = "HOP_ZIP_MAX_ENTRY_SIZE";
/**
@@ -679,7 +797,7 @@ public class Const {
*
* @see #HOP_ZIP_MAX_ENTRY_SIZE
* @see #HOP_ZIP_MAX_ENTRY_SIZE_DEFAULT_STRING
- */
+ */
public static final Long HOP_ZIP_MAX_ENTRY_SIZE_DEFAULT = 0xFFFFFFFFL;
/**
@@ -688,7 +806,7 @@ public class Const {
* @see #HOP_ZIP_MAX_ENTRY_SIZE
* @see #HOP_ZIP_MAX_ENTRY_SIZE_DEFAULT
*/
- @Variable(description="")
+ @Variable(description = "")
public static final String HOP_ZIP_MAX_ENTRY_SIZE_DEFAULT_STRING =
String.valueOf(HOP_ZIP_MAX_ENTRY_SIZE_DEFAULT);
@@ -702,7 +820,9 @@ public class Const {
* @see #HOP_ZIP_MAX_TEXT_SIZE_DEFAULT
* @see #HOP_ZIP_MAX_TEXT_SIZE_DEFAULT_STRING
*/
- @Variable(description="A variable to configure the maximum number of characters of text that are extracted before an exception is thrown during extracting text from documents")
+ @Variable(
+ description =
+ "A variable to configure the maximum number of characters of text that are extracted before an exception is thrown during extracting text from documents")
public static final String HOP_ZIP_MAX_TEXT_SIZE = "HOP_ZIP_MAX_TEXT_SIZE";
/**
@@ -719,7 +839,7 @@ public class Const {
* @see #HOP_ZIP_MAX_TEXT_SIZE
* @see #HOP_ZIP_MAX_TEXT_SIZE_DEFAULT
*/
- @Variable(description="")
+ @Variable(description = "")
public static final String HOP_ZIP_MAX_TEXT_SIZE_DEFAULT_STRING =
String.valueOf(HOP_ZIP_MAX_TEXT_SIZE_DEFAULT);
@@ -727,7 +847,9 @@ public class Const {
* This is the name of the variable which when set should contains the path to a file which will
* be included in the serialization of pipelines and workflows.
*/
- @Variable(description="This is the name of the variable which when set should contains the path to a file which will be included in the serialization of pipelines and workflows")
+ @Variable(
+ description =
+ "This is the name of the variable which when set should contains the path to a file which will be included in the serialization of pipelines and workflows")
public static final String HOP_LICENSE_HEADER_FILE = "HOP_LICENSE_HEADER_FILE";
/** The i18n prefix to signal that this is a String in the format: i18n:package:key */
@@ -738,6 +860,17 @@ public class Const {
*/
public static final String CONNECTION_GROUP = "CONNECTION_GROUP";
+ /**
+ * This is the default wait time used for DynamicWaitTimes can be overwritten by a runtime
+ * configuration
+ */
+ @Variable(
+ scope = VariableScope.ENGINE,
+ value = "20",
+ description = "This is the default polling frequency for the transforms input buffer (in ms)")
+ public static final String HOP_DEFAULT_BUFFER_POLLING_WAITTIME =
+ "HOP_DEFAULT_BUFFER_POLLING_WAITTIME";
+
/**
* rounds double f to any number of places after decimal point Does arithmetic using BigDecimal
* class to avoid integer overflow while rounding
@@ -1130,7 +1263,9 @@ public class Const {
return true;
}
- /** @return the operating system hop runs on */
+ /**
+ * @return the operating system hop runs on
+ */
public static String getSystemOs() {
return System.getProperty("os.name");
}
@@ -1188,22 +1323,30 @@ public class Const {
}
}
- /** @return True if the OS is a Windows derivate. */
+ /**
+ * @return True if the OS is a Windows derivate.
+ */
public static boolean isWindows() {
return getHopPlatformOs().startsWith("Windows") || getSystemOs().startsWith("Windows");
}
- /** @return True if the OS is a Linux derivate. */
+ /**
+ * @return True if the OS is a Linux derivate.
+ */
public static boolean isLinux() {
return getHopPlatformOs().startsWith("Linux") || getSystemOs().startsWith("Linux");
}
- /** @return True if the OS is an OSX derivate. */
+ /**
+ * @return True if the OS is an OSX derivate.
+ */
public static boolean isOSX() {
return getHopPlatformOs().startsWith("Darwin") || getSystemOs().toUpperCase().contains("OS X");
}
- /** @return True if KDE is in use. */
+ /**
+ * @return True if KDE is in use.
+ */
public static boolean isKDE() {
return StringUtils.isNotBlank(System.getenv("KDE_SESSION_VERSION"));
}
@@ -2192,7 +2335,9 @@ public class Const {
return numberFormats;
}
- /** @return An array of all default conversion formats, to be used in dialogs etc. */
+ /**
+ * @return An array of all default conversion formats, to be used in dialogs etc.
+ */
public static String[] getConversionFormats() {
String[] dats = Const.getDateFormats();
String[] nums = Const.getNumberFormats();
diff --git a/docs/hop-user-manual/modules/ROOT/pages/pipeline/pipeline-run-configurations/native-local-pipeline-engine.adoc b/docs/hop-user-manual/modules/ROOT/pages/pipeline/pipeline-run-configurations/native-local-pipeline-engine.adoc
index fe519324d2..7a7e625e6e 100644
--- a/docs/hop-user-manual/modules/ROOT/pages/pipeline/pipeline-run-configurations/native-local-pipeline-engine.adoc
+++ b/docs/hop-user-manual/modules/ROOT/pages/pipeline/pipeline-run-configurations/native-local-pipeline-engine.adoc
@@ -57,6 +57,10 @@ If a row does not have the same layout as the first row, an error is generated a
|The number of rows to return as feedback.
|50.000
+|Wait time buffer check (ms)
+|This indicates the polling frequency when there are no rows in the input buffer of a transform, a lower value will cause higher CPU load when the pipeline has many idle transforms.
+|20
+
|Sample type while running in the GUI
|Here you can specify which rows you want to see when you click on the little grid icons on the transform icons during a pipeline execution.
You can choose: `None`: do not show, `First` : show the first rows, `Last` : show the last rows or `Random`: a random selection of rows (reservoir sampling)
diff --git a/engine/src/main/java/org/apache/hop/pipeline/engines/local/LocalPipelineRunConfiguration.java b/engine/src/main/java/org/apache/hop/pipeline/engines/local/LocalPipelineRunConfiguration.java
index 42f1aaf6ef..14ebcf3e53 100644
--- a/engine/src/main/java/org/apache/hop/pipeline/engines/local/LocalPipelineRunConfiguration.java
+++ b/engine/src/main/java/org/apache/hop/pipeline/engines/local/LocalPipelineRunConfiguration.java
@@ -22,6 +22,7 @@ import org.apache.hop.core.gui.plugin.GuiElementType;
import org.apache.hop.core.gui.plugin.GuiPlugin;
import org.apache.hop.core.gui.plugin.GuiWidgetElement;
import org.apache.hop.core.logging.ILogChannel;
+import org.apache.hop.core.util.EnvUtil;
import org.apache.hop.metadata.api.HopMetadataProperty;
import org.apache.hop.metadata.api.IHopMetadataProvider;
import org.apache.hop.pipeline.config.IPipelineEngineRunConfiguration;
@@ -99,9 +100,20 @@ public class LocalPipelineRunConfiguration extends EmptyPipelineRunConfiguration
/** The feedback size. */
@GuiWidgetElement(
- id = "sampleTypeInGui",
+ id = "waitTime",
order = "070",
parentId = PipelineRunConfiguration.GUI_PLUGIN_ELEMENT_PARENT_ID,
+ type = GuiElementType.TEXT,
+ label =
+ "i18n:org.apache.hop.ui.pipeline.config:PipelineRunConfigurationDialog.WaitTime.Label")
+ @HopMetadataProperty(key = "wait_time")
+ protected String waitTime;
+
+ /** The feedback size. */
+ @GuiWidgetElement(
+ id = "sampleTypeInGui",
+ order = "080",
+ parentId = PipelineRunConfiguration.GUI_PLUGIN_ELEMENT_PARENT_ID,
type = GuiElementType.COMBO,
label =
"i18n:org.apache.hop.ui.pipeline.config:PipelineRunConfigurationDialog.SampleTypeInGui.Label",
@@ -112,7 +124,7 @@ public class LocalPipelineRunConfiguration extends EmptyPipelineRunConfiguration
/** The feedback size. */
@GuiWidgetElement(
id = "sampleSize",
- order = "080",
+ order = "090",
parentId = PipelineRunConfiguration.GUI_PLUGIN_ELEMENT_PARENT_ID,
type = GuiElementType.TEXT,
label =
@@ -122,7 +134,7 @@ public class LocalPipelineRunConfiguration extends EmptyPipelineRunConfiguration
@GuiWidgetElement(
id = "transactional",
- order = "090",
+ order = "100",
parentId = PipelineRunConfiguration.GUI_PLUGIN_ELEMENT_PARENT_ID,
type = GuiElementType.CHECKBOX,
label =
@@ -144,6 +156,7 @@ public class LocalPipelineRunConfiguration extends EmptyPipelineRunConfiguration
this.rowSetSize = Integer.toString(Const.ROWS_IN_ROWSET);
this.feedbackShown = false;
this.feedbackSize = Integer.toString(Const.ROWS_UPDATE);
+ this.waitTime = EnvUtil.getSystemProperty(Const.HOP_DEFAULT_BUFFER_POLLING_WAITTIME, "20");
this.sampleTypeInGui = SampleType.Last.name();
this.sampleSize = "100";
this.transactional = false;
@@ -154,6 +167,7 @@ public class LocalPipelineRunConfiguration extends EmptyPipelineRunConfiguration
this.rowSetSize = config.rowSetSize;
this.feedbackShown = config.feedbackShown;
this.feedbackSize = config.feedbackSize;
+ this.waitTime = config.waitTime;
this.safeModeEnabled = config.safeModeEnabled;
this.gatheringMetrics = config.gatheringMetrics;
this.sortingTransformsTopologically = config.sortingTransformsTopologically;
@@ -183,7 +197,9 @@ public class LocalPipelineRunConfiguration extends EmptyPipelineRunConfiguration
return rowSetSize;
}
- /** @param rowSetSize The rowSetSize to set */
+ /**
+ * @param rowSetSize The rowSetSize to set
+ */
public void setRowSetSize(String rowSetSize) {
this.rowSetSize = rowSetSize;
}
@@ -197,7 +213,9 @@ public class LocalPipelineRunConfiguration extends EmptyPipelineRunConfiguration
return safeModeEnabled;
}
- /** @param safeModeEnabled The safeModeEnabled to set */
+ /**
+ * @param safeModeEnabled The safeModeEnabled to set
+ */
public void setSafeModeEnabled(boolean safeModeEnabled) {
this.safeModeEnabled = safeModeEnabled;
}
@@ -211,7 +229,9 @@ public class LocalPipelineRunConfiguration extends EmptyPipelineRunConfiguration
return gatheringMetrics;
}
- /** @param gatheringMetrics The gatheringMetrics to set */
+ /**
+ * @param gatheringMetrics The gatheringMetrics to set
+ */
public void setGatheringMetrics(boolean gatheringMetrics) {
this.gatheringMetrics = gatheringMetrics;
}
@@ -225,7 +245,9 @@ public class LocalPipelineRunConfiguration extends EmptyPipelineRunConfiguration
return sortingTransformsTopologically;
}
- /** @param sortingTransformsTopologically The sortingTransformsTopologically to set */
+ /**
+ * @param sortingTransformsTopologically The sortingTransformsTopologically to set
+ */
public void setSortingTransformsTopologically(boolean sortingTransformsTopologically) {
this.sortingTransformsTopologically = sortingTransformsTopologically;
}
@@ -239,7 +261,9 @@ public class LocalPipelineRunConfiguration extends EmptyPipelineRunConfiguration
return feedbackShown;
}
- /** @param feedbackShown The feedbackShown to set */
+ /**
+ * @param feedbackShown The feedbackShown to set
+ */
public void setFeedbackShown(boolean feedbackShown) {
this.feedbackShown = feedbackShown;
}
@@ -253,11 +277,29 @@ public class LocalPipelineRunConfiguration extends EmptyPipelineRunConfiguration
return feedbackSize;
}
- /** @param feedbackSize The feedbackSize to set */
+ /**
+ * @param feedbackSize The feedbackSize to set
+ */
public void setFeedbackSize(String feedbackSize) {
this.feedbackSize = feedbackSize;
}
+ /**
+ * Gets waitTime
+ *
+ * @return value of the waitTime
+ */
+ public String getWaitTime() {
+ return waitTime;
+ }
+
+ /**
+ * @param waitTime The waitTime for buffer check
+ */
+ public void setWaitTime(String waitTime) {
+ this.waitTime = waitTime;
+ }
+
/**
* Gets sampleTypeInGui
*
@@ -267,7 +309,9 @@ public class LocalPipelineRunConfiguration extends EmptyPipelineRunConfiguration
return sampleTypeInGui;
}
- /** @param sampleTypeInGui The sampleTypeInGui to set */
+ /**
+ * @param sampleTypeInGui The sampleTypeInGui to set
+ */
public void setSampleTypeInGui(String sampleTypeInGui) {
this.sampleTypeInGui = sampleTypeInGui;
}
@@ -281,7 +325,9 @@ public class LocalPipelineRunConfiguration extends EmptyPipelineRunConfiguration
return sampleSize;
}
- /** @param sampleSize The sampleSize to set */
+ /**
+ * @param sampleSize The sampleSize to set
+ */
public void setSampleSize(String sampleSize) {
this.sampleSize = sampleSize;
}
@@ -295,7 +341,9 @@ public class LocalPipelineRunConfiguration extends EmptyPipelineRunConfiguration
return transactional;
}
- /** @param transactional The transactional to set */
+ /**
+ * @param transactional The transactional to set
+ */
public void setTransactional(boolean transactional) {
this.transactional = transactional;
}
diff --git a/engine/src/main/java/org/apache/hop/pipeline/transform/BaseTransform.java b/engine/src/main/java/org/apache/hop/pipeline/transform/BaseTransform.java
index f212f6e0da..fe5306e069 100644
--- a/engine/src/main/java/org/apache/hop/pipeline/transform/BaseTransform.java
+++ b/engine/src/main/java/org/apache/hop/pipeline/transform/BaseTransform.java
@@ -32,6 +32,7 @@ import org.apache.hop.core.row.IValueMeta;
import org.apache.hop.core.row.RowDataUtil;
import org.apache.hop.core.row.value.ValueMetaBase;
import org.apache.hop.core.row.value.ValueMetaString;
+import org.apache.hop.core.util.EnvUtil;
import org.apache.hop.core.util.Utils;
import org.apache.hop.core.variables.IVariables;
import org.apache.hop.core.variables.Variables;
@@ -43,6 +44,7 @@ import org.apache.hop.pipeline.PipelineMeta;
import org.apache.hop.pipeline.engine.EngineComponent.ComponentExecutionStatus;
import org.apache.hop.pipeline.engine.IEngineComponent;
import org.apache.hop.pipeline.engine.IPipelineEngine;
+import org.apache.hop.pipeline.engines.local.LocalPipelineRunConfiguration;
import java.io.Closeable;
import java.io.IOException;
@@ -115,12 +117,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
* </ul>
*/
public class BaseTransform<Meta extends ITransformMeta, Data extends ITransformData>
- implements
- ITransform,
- IVariables,
- ILoggingObject,
- IExtensionData,
- IEngineComponent {
+ implements ITransform, IVariables, ILoggingObject, IExtensionData, IEngineComponent {
private static final Class<?> PKG = BaseTransform.class; // For Translator
@@ -461,9 +458,9 @@ public class BaseTransform<Meta extends ITransformMeta, Data extends ITransformD
setVariable(Const.INTERNAL_VARIABLE_TRANSFORM_COPYNR, Integer.toString(copyNr));
// BACKLOG-18004
- allowEmptyFieldNamesAndTypes = ValueMetaBase.convertStringToBoolean(
- System.getProperties()
- .getProperty(Const.HOP_ALLOW_EMPTY_FIELD_NAMES_AND_TYPES, "N"));
+ allowEmptyFieldNamesAndTypes =
+ ValueMetaBase.convertStringToBoolean(
+ System.getProperties().getProperty(Const.HOP_ALLOW_EMPTY_FIELD_NAMES_AND_TYPES, "N"));
// Getting ans setting the error handling values
// first, get the transform meta
@@ -579,7 +576,9 @@ public class BaseTransform<Meta extends ITransformMeta, Data extends ITransformD
copyNr = cop;
}
- /** @return The transforms copy number (default 0) */
+ /**
+ * @return The transforms copy number (default 0)
+ */
@Override
public int getCopy() {
return copyNr;
@@ -907,7 +906,9 @@ public class BaseTransform<Meta extends ITransformMeta, Data extends ITransformD
return getStatus().getDescription();
}
- /** @return Returns the transformMeta. */
+ /**
+ * @return Returns the transformMeta.
+ */
@Override
public Meta getMeta() {
return meta;
@@ -923,18 +924,24 @@ public class BaseTransform<Meta extends ITransformMeta, Data extends ITransformD
return data;
}
- /** @return Returns the transformMeta. */
+ /**
+ * @return Returns the transformMeta.
+ */
@Override
public TransformMeta getTransformMeta() {
return transformMeta;
}
- /** @return Returns the pipelineMeta. */
+ /**
+ * @return Returns the pipelineMeta.
+ */
public PipelineMeta getPipelineMeta() {
return pipelineMeta;
}
- /** @return Returns the pipeline. */
+ /**
+ * @return Returns the pipeline.
+ */
@Override
public IPipelineEngine<PipelineMeta> getPipeline() {
return pipeline;
@@ -1011,8 +1018,8 @@ public class BaseTransform<Meta extends ITransformMeta, Data extends ITransformD
} catch (InterruptedException e) {
// Ignore
}
- //wait 3s max
- if(counter >= 3) {
+ // wait 3s max
+ if (counter >= 3) {
break;
}
}
@@ -1491,7 +1498,6 @@ public class BaseTransform<Meta extends ITransformMeta, Data extends ITransformD
return row;
}
-
private Object[] handleGetRow() throws HopException {
// Are we pausing the transform? If so, stall forever...
@@ -1573,7 +1579,17 @@ public class BaseTransform<Meta extends ITransformMeta, Data extends ITransformD
// We can use timeouts to switch from one to another...
//
if (waitingTime == null) {
- waitingTime = DynamicWaitTimes.build(inputRowSets, this::getCurrentInputRowSetNr);
+ Integer waitTime =
+ Const.toInt(EnvUtil.getSystemProperty(Const.HOP_DEFAULT_BUFFER_POLLING_WAITTIME), 20);
+ ;
+ if (pipeline.getPipelineRunConfiguration().getEngineRunConfiguration()
+ instanceof LocalPipelineRunConfiguration) {
+ LocalPipelineRunConfiguration runconfig =
+ (LocalPipelineRunConfiguration)
+ pipeline.getPipelineRunConfiguration().getEngineRunConfiguration();
+ waitTime = Const.toInt(runconfig.getWaitTime(), waitTime);
+ }
+ waitingTime = DynamicWaitTimes.build(inputRowSets, this::getCurrentInputRowSetNr, waitTime);
}
while (row == null && !isStopped()) {
// Get a row from the input in row set ...
@@ -1891,8 +1907,19 @@ public class BaseTransform<Meta extends ITransformMeta, Data extends ITransformD
//
rowData = rowSet.getRow();
if (rowData == null) {
+ Integer waitTime = 20;
+ if (pipeline.getPipelineRunConfiguration().getEngineRunConfiguration()
+ instanceof LocalPipelineRunConfiguration) {
+ LocalPipelineRunConfiguration runconfig =
+ (LocalPipelineRunConfiguration)
+ pipeline.getPipelineRunConfiguration().getEngineRunConfiguration();
+ waitTime = Integer.parseInt(runconfig.getWaitTime());
+ }
+ if (waitTime == null) waitTime = 20;
+
if (waitingTime == null) {
- waitingTime = DynamicWaitTimes.build(inputRowSets, this::getCurrentInputRowSetNr);
+ waitingTime =
+ DynamicWaitTimes.build(inputRowSets, this::getCurrentInputRowSetNr, waitTime);
}
// Must release the read lock before acquisition of the write lock to prevent deadlocks.
//
@@ -2882,7 +2909,9 @@ public class BaseTransform<Meta extends ITransformMeta, Data extends ITransformD
return null;
}
- /** @return Returns the inputRowSets. */
+ /**
+ * @return Returns the inputRowSets.
+ */
@Override
public List<IRowSet> getInputRowSets() {
inputRowSetsLock.readLock().lock();
@@ -2947,7 +2976,9 @@ public class BaseTransform<Meta extends ITransformMeta, Data extends ITransformD
}
}
- /** @param inputRowSets The inputRowSets to set. */
+ /**
+ * @param inputRowSets The inputRowSets to set.
+ */
public void setInputRowSets(List<IRowSet> inputRowSets) {
inputRowSetsLock.writeLock().lock();
try {
@@ -2957,7 +2988,9 @@ public class BaseTransform<Meta extends ITransformMeta, Data extends ITransformD
}
}
- /** @return Returns the outputRowSets. */
+ /**
+ * @return Returns the outputRowSets.
+ */
@Override
public List<IRowSet> getOutputRowSets() {
outputRowSetsLock.readLock().lock();
@@ -2987,7 +3020,9 @@ public class BaseTransform<Meta extends ITransformMeta, Data extends ITransformD
}
}
- /** @param outputRowSets The outputRowSets to set. */
+ /**
+ * @param outputRowSets The outputRowSets to set.
+ */
public void setOutputRowSets(List<IRowSet> outputRowSets) {
outputRowSetsLock.writeLock().lock();
try {
@@ -2997,12 +3032,16 @@ public class BaseTransform<Meta extends ITransformMeta, Data extends ITransformD
}
}
- /** @return Returns the distributed. */
+ /**
+ * @return Returns the distributed.
+ */
public boolean isDistributed() {
return distributed;
}
- /** @param distributed The distributed to set. */
+ /**
+ * @param distributed The distributed to set.
+ */
public void setDistributed(boolean distributed) {
this.distributed = distributed;
}
@@ -3119,46 +3158,62 @@ public class BaseTransform<Meta extends ITransformMeta, Data extends ITransformD
}
}
- /** @return the partitionId */
+ /**
+ * @return the partitionId
+ */
@Override
public String getPartitionId() {
return partitionId;
}
- /** @param partitionId the partitionId to set */
+ /**
+ * @param partitionId the partitionId to set
+ */
@Override
public void setPartitionId(String partitionId) {
this.partitionId = partitionId;
}
- /** @return the partitionTargets */
+ /**
+ * @return the partitionTargets
+ */
public Map<String, BlockingRowSet> getPartitionTargets() {
return partitionTargets;
}
- /** @param partitionTargets the partitionTargets to set */
+ /**
+ * @param partitionTargets the partitionTargets to set
+ */
public void setPartitionTargets(Map<String, BlockingRowSet> partitionTargets) {
this.partitionTargets = partitionTargets;
}
- /** @return the repartitioning type */
+ /**
+ * @return the repartitioning type
+ */
public int getRepartitioning() {
return repartitioning;
}
- /** @param repartitioning the repartitioning type to set */
+ /**
+ * @param repartitioning the repartitioning type to set
+ */
@Override
public void setRepartitioning(int repartitioning) {
this.repartitioning = repartitioning;
}
- /** @return the partitioned */
+ /**
+ * @return the partitioned
+ */
@Override
public boolean isPartitioned() {
return partitioned;
}
- /** @param partitioned the partitioned to set */
+ /**
+ * @param partitioned the partitioned to set
+ */
@Override
public void setPartitioned(boolean partitioned) {
this.partitioned = partitioned;
@@ -3177,32 +3232,44 @@ public class BaseTransform<Meta extends ITransformMeta, Data extends ITransformD
&& (lines % getPipeline().getFeedbackSize()) == 0;
}
- /** @return the rowMeta */
+ /**
+ * @return the rowMeta
+ */
public IRowMeta getInputRowMeta() {
return inputRowMeta;
}
- /** @param rowMeta the rowMeta to set */
+ /**
+ * @param rowMeta the rowMeta to set
+ */
public void setInputRowMeta(IRowMeta rowMeta) {
this.inputRowMeta = rowMeta;
}
- /** @return the errorRowMeta */
+ /**
+ * @return the errorRowMeta
+ */
public IRowMeta getErrorRowMeta() {
return errorRowMeta;
}
- /** @param errorRowMeta the errorRowMeta to set */
+ /**
+ * @param errorRowMeta the errorRowMeta to set
+ */
public void setErrorRowMeta(IRowMeta errorRowMeta) {
this.errorRowMeta = errorRowMeta;
}
- /** @return the previewRowMeta */
+ /**
+ * @return the previewRowMeta
+ */
public IRowMeta getPreviewRowMeta() {
return previewRowMeta;
}
- /** @param previewRowMeta the previewRowMeta to set */
+ /**
+ * @param previewRowMeta the previewRowMeta to set
+ */
public void setPreviewRowMeta(IRowMeta previewRowMeta) {
this.previewRowMeta = previewRowMeta;
}
@@ -3424,7 +3491,9 @@ public class BaseTransform<Meta extends ITransformMeta, Data extends ITransformD
return transformFinishedListeners;
}
- /** @param transformFinishedListeners The transformFinishedListeners to set */
+ /**
+ * @param transformFinishedListeners The transformFinishedListeners to set
+ */
public void setTransformFinishedListeners(
List<ITransformFinishedListener> transformFinishedListeners) {
this.transformFinishedListeners = transformFinishedListeners;
@@ -3439,7 +3508,9 @@ public class BaseTransform<Meta extends ITransformMeta, Data extends ITransformD
return transformStartedListeners;
}
- /** @param transformStartedListeners The transformStartedListeners to set */
+ /**
+ * @param transformStartedListeners The transformStartedListeners to set
+ */
public void setTransformStartedListeners(
List<ITransformStartedListener> transformStartedListeners) {
this.transformStartedListeners = transformStartedListeners;
@@ -3485,7 +3556,9 @@ public class BaseTransform<Meta extends ITransformMeta, Data extends ITransformD
return copyNr;
}
- /** @param copyNr The copyNr to set */
+ /**
+ * @param copyNr The copyNr to set
+ */
public void setCopyNr(int copyNr) {
this.copyNr = copyNr;
}
@@ -3725,7 +3798,9 @@ public class BaseTransform<Meta extends ITransformMeta, Data extends ITransformD
return initStartDate;
}
- /** @param initStartDate The initStartDate to set */
+ /**
+ * @param initStartDate The initStartDate to set
+ */
@Override
public void setInitStartDate(Date initStartDate) {
this.initStartDate = initStartDate;
@@ -3741,7 +3816,9 @@ public class BaseTransform<Meta extends ITransformMeta, Data extends ITransformD
return executionStartDate;
}
- /** @param executionStartDate The executionStartDate to set */
+ /**
+ * @param executionStartDate The executionStartDate to set
+ */
@Override
public void setExecutionStartDate(Date executionStartDate) {
this.executionStartDate = executionStartDate;
@@ -3757,7 +3834,9 @@ public class BaseTransform<Meta extends ITransformMeta, Data extends ITransformD
return firstRowReadDate;
}
- /** @param firstRowReadDate The firstRowReadDate to set */
+ /**
+ * @param firstRowReadDate The firstRowReadDate to set
+ */
@Override
public void setFirstRowReadDate(Date firstRowReadDate) {
this.firstRowReadDate = firstRowReadDate;
@@ -3773,7 +3852,9 @@ public class BaseTransform<Meta extends ITransformMeta, Data extends ITransformD
return lastRowWrittenDate;
}
- /** @param lastRowWrittenDate The lastRowWrittenDate to set */
+ /**
+ * @param lastRowWrittenDate The lastRowWrittenDate to set
+ */
@Override
public void setLastRowWrittenDate(Date lastRowWrittenDate) {
this.lastRowWrittenDate = lastRowWrittenDate;
@@ -3789,7 +3870,9 @@ public class BaseTransform<Meta extends ITransformMeta, Data extends ITransformD
return executionEndDate;
}
- /** @param executionEndDate The executionEndDate to set */
+ /**
+ * @param executionEndDate The executionEndDate to set
+ */
@Override
public void setExecutionEndDate(Date executionEndDate) {
this.executionEndDate = executionEndDate;
diff --git a/engine/src/main/java/org/apache/hop/pipeline/transform/DynamicWaitTimes.java b/engine/src/main/java/org/apache/hop/pipeline/transform/DynamicWaitTimes.java
index 5be9df6669..687f8a8ad5 100644
--- a/engine/src/main/java/org/apache/hop/pipeline/transform/DynamicWaitTimes.java
+++ b/engine/src/main/java/org/apache/hop/pipeline/transform/DynamicWaitTimes.java
@@ -26,18 +26,22 @@ import java.util.function.Supplier;
final class DynamicWaitTimes {
- static final long MAX_TIMEOUT = 1000;
- static SingleStreamStatus build(List<IRowSet> rowSets, Supplier<Integer> supplier) {
+ static SingleStreamStatus build(List<IRowSet> rowSets, Supplier<Integer> supplier, Integer waitTime) {
if (rowSets.size() == 1) {
- return new SingleStreamStatus();
+ return new SingleStreamStatus(waitTime);
}
- return new MultiStreamStatus(new ArrayList<>(rowSets), supplier);
+ return new MultiStreamStatus(new ArrayList<>(rowSets), supplier, waitTime);
}
static class SingleStreamStatus {
protected boolean active = true;
private long interval = 1;
+ private long waitTime;
+
+ SingleStreamStatus(Integer waitTime) {
+ this.waitTime = waitTime;
+ }
public long get() {
return interval;
@@ -50,12 +54,12 @@ final class DynamicWaitTimes {
public void adjust(boolean timeout, IRowSet nextIfExist) {
if (allowAdjust() && timeout) {
- if (interval == MAX_TIMEOUT) {
+ if (interval == waitTime) {
active = false;
}
interval = interval * 2;
- if (interval > MAX_TIMEOUT) {
- interval = MAX_TIMEOUT;
+ if (interval > waitTime) {
+ interval = waitTime;
}
}
}
@@ -75,12 +79,13 @@ final class DynamicWaitTimes {
private final List<SingleStreamStatus> statusList;
private final Supplier<Integer> supplier;
- MultiStreamStatus(List<IRowSet> rowSets, Supplier<Integer> supplier) {
+ MultiStreamStatus(List<IRowSet> rowSets, Supplier<Integer> supplier, Integer waitTime) {
+ super(waitTime);
this.streamList = rowSets;
this.supplier = supplier;
this.statusList = new ArrayList<>(rowSets.size());
for (int i = 0; i < rowSets.size(); i++) {
- statusList.add(new SingleStreamStatus());
+ statusList.add(new SingleStreamStatus(waitTime));
}
}
diff --git a/engine/src/test/java/org/apache/hop/pipeline/transform/DynamicWaitTimesTest.java b/engine/src/test/java/org/apache/hop/pipeline/transform/DynamicWaitTimesTest.java
index bcf37143e4..83eb1ed417 100644
--- a/engine/src/test/java/org/apache/hop/pipeline/transform/DynamicWaitTimesTest.java
+++ b/engine/src/test/java/org/apache/hop/pipeline/transform/DynamicWaitTimesTest.java
@@ -34,21 +34,21 @@ public class DynamicWaitTimesTest extends TestCase {
public void testSingleStreamStatus() {
IRowSet rowSet = new BlockingRowSet(3);
- status = DynamicWaitTimes.build(Collections.singletonList(rowSet), () -> 0);
+ status = DynamicWaitTimes.build(Collections.singletonList(rowSet), () -> 0, 20);
assertEquals(1, status.get());
status.adjust(true, rowSet);
assertEquals(2, status.get());
for (int i = 0; i < 10; i++) {
status.adjust(true, rowSet);
}
- assertEquals(DynamicWaitTimes.MAX_TIMEOUT, status.get());
+ assertEquals(20, status.get());
}
public void testMultiStreamStatus() {
List<IRowSet> rowSetList =
new ArrayList<>(
Arrays.asList(new BlockingRowSet(1), new BlockingRowSet(2), new BlockingRowSet(7)));
- status = DynamicWaitTimes.build(rowSetList, () -> activeStreamIndex.get());
+ status = DynamicWaitTimes.build(rowSetList, () -> activeStreamIndex.get(),20);
for (IRowSet iRowSet : rowSetList) {
status.adjust(false, iRowSet);
assertEquals(1, status.get());
diff --git a/integration-tests/transforms/metadata/pipeline-run-configuration/local.json b/integration-tests/transforms/metadata/pipeline-run-configuration/local.json
index d577d578d7..e5c96a6aaf 100644
--- a/integration-tests/transforms/metadata/pipeline-run-configuration/local.json
+++ b/integration-tests/transforms/metadata/pipeline-run-configuration/local.json
@@ -2,13 +2,15 @@
"engineRunConfiguration": {
"Local": {
"feedback_size": "50000",
+ "wait_time": "20",
"sample_size": "100",
"sample_type_in_gui": "First",
"rowset_size": "10000",
"safe_mode": false,
"show_feedback": false,
"topo_sort": false,
- "gather_metrics": false
+ "gather_metrics": false,
+ "transactional": false
}
},
"configurationVariables": [],
diff --git a/plugins/transforms/xml/src/test/java/org/apache/hop/pipeline/transforms/xml/addxml/AddXmlTest.java b/plugins/transforms/xml/src/test/java/org/apache/hop/pipeline/transforms/xml/addxml/AddXmlTest.java
index 7dde0d4319..daf8a1265a 100644
--- a/plugins/transforms/xml/src/test/java/org/apache/hop/pipeline/transforms/xml/addxml/AddXmlTest.java
+++ b/plugins/transforms/xml/src/test/java/org/apache/hop/pipeline/transforms/xml/addxml/AddXmlTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.hop.pipeline.transforms.xml.addxml;
+import org.apache.hop.core.HopEnvironment;
import org.apache.hop.core.IRowSet;
import org.apache.hop.core.exception.HopException;
import org.apache.hop.core.logging.ILoggingObject;
@@ -38,6 +39,7 @@ public class AddXmlTest {
@Before
public void setup() throws Exception {
+ HopEnvironment.init();
XmlField field = mock(XmlField.class);
when(field.getElementName()).thenReturn("ADDXML_TEST");
when(field.isAttribute()).thenReturn(true);
diff --git a/ui/src/main/resources/org/apache/hop/ui/pipeline/config/messages/messages_en_US.properties b/ui/src/main/resources/org/apache/hop/ui/pipeline/config/messages/messages_en_US.properties
index 3c744d28a8..30a0693452 100644
--- a/ui/src/main/resources/org/apache/hop/ui/pipeline/config/messages/messages_en_US.properties
+++ b/ui/src/main/resources/org/apache/hop/ui/pipeline/config/messages/messages_en_US.properties
@@ -29,6 +29,7 @@ PipelineRunConfigurationDialog.GatheringMetrics.Label=Collect metrics
PipelineRunConfigurationDialog.SortTransformsTopologically.Label=Sort transforms
PipelineRunConfigurationDialog.FeedbackShown.Label=Log rows feedback
PipelineRunConfigurationDialog.FeedbackSize.Label=Feedback size in rows
+PipelineRunConfigurationDialog.WaitTime.Label=Wait time for buffer check (ms)
PipelineRunConfigurationDialog.SampleTypeInGui.Label=Sample type while running in the GUI
PipelineRunConfigurationDialog.SampleSize.Label=Number of rows to sample in the GUI
PipelineRunConfigurationDialog.HopServer.Label=Hop server