You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by az...@apache.org on 2020/04/20 09:41:24 UTC
[flink] branch flip116 updated (d433ea1c -> 28b13e7)
This is an automated email from the ASF dual-hosted git repository.
azagrebin pushed a change to branch flip116
in repository https://gitbox.apache.org/repos/asf/flink.git.
discard d433ea1c fixup! [FLINK-16745][k8s] Start Kubernetes JM with FLIP-116 JVM memory args
discard e299488 change type to cron
omit 74f7619 [FLINK-16746][conf] Deprecate jobmanager.heap.size
omit 9e0a523 [FLINK-16746] Add default 'jobmanager.memory.process.size: 1472m' to flink-conf.yaml
omit e3fdf98 [hotfix] Remove 'TaskManager' from MemoryBackwardsCompatibilityUtils#getLegacyTaskManagerHeapMemoryIfExplicitlyConfigured
omit cf91b68 [FLINK-16745][coordination] Remove unused container cutoff
omit 8adb309 [FLINK-16745][k8s] Start Kubernetes JM with FLIP-116 JVM memory args
omit 0c84a68 [FLINK-16745][yarn] Start Yarn JM with FLIP-116 JVM memory args
omit 7d31edc [FLINK-16745][clients] Parse JobManagerProcessSpec from Configuration into ClusterSpecification
omit 1504abb [FLINK-16742][runtime] Refactor BashJavaUtils, move to separated package and extract configuration loading logics.
omit 11775b3 [FLINK-16472][dist] Remove unused memory configuration logics from config.sh.
omit 340fde2 [FLINK-16742][dist] Extend and use BashJavaUtils for JM memory configuration in start-up scripts.
omit b73c996 [FLINK-16742][runtime] Ignore unknown command line options for BashJavaUtils.
omit d2e199d [FLINK-16742][dist] config.sh does not assume BashJavaUtils always return two lines of results.
omit fd6174c [hotfix][runtime] Code clean-ups in BashJavaUtils and its test cases.
omit 295449f [FLINK-16874] Respect the dynamic options when calculating memory options in taskmanager.sh
omit 5fce7df [hotfix] Fix the code style in BashJavaUtilsITCase
add d3fa2e5 Revert "[FLINK-16945][checkpointing] Execute CheckpointFailureManager.FailJobCallback directly in main thread executor"
add cfe5a27 Revert "[hotfix] Make the checkpoint resuming e2e case pass by increasing the retained checkpoints number"
add 5d5a29b Revert "[FLINK-14971][checkpointing] Remove lock of PendingCheckpoint and introduce IO lock for PendingCheckpoint"
add 214896a Revert "[FLINK-14971][checkpointing] Remove coordinator-wide lock of CheckpointCoordinator"
add 6adbe94 Revert "[FLINK-14971][checkpointing] Make CompletedCheckpointStore thread-safe to avoid synchronization outside"
add e217a8c Revert "[hotfix] Harden ResumeCheckpointManuallyITCase"
add 2813be7 Revert "[FLINK-14971][checkpointing] Introduce main thread executor in CheckpointCoordinator to execute all non-IO operations instead of the timer thread"
add ffa4475 Revert "[FLINK-14971][checkpointing] Handle ack/declined message of checkpoint"
add 4ed8bb8 [FLINK-16308] [docs] Fix Maven download links for SQL connectors.
add 4799d2e [minor] Add ES7 to list of sql connector downloads
add 146b34e [FLINK-17052] Extract the Plan Generation logic from ExecutionEnvironment
add 7b3db65 [FLINK-16967][metrics][slf4j] Add Slf4jReporterFactory
add 5421071 [FLINK-16962][metrics][datadog] Add DatadogHttpReporterFactory
add 9f113e4 Revert "[FLINK-17118][python] Add Cython support for primitive data types (#11718)"
add 04337dd [FLINK-17211] Enable JUnit 4 tests in flink-sql-parser module
add 69cfc11 [FLINK-16921][e2e] Use kubectl instead of curl to wait for rest endpoint up
add 9b50470 [FLINK-16921][e2e] Set service exposed type to NodePort explicitly
add b7f16a4 [FLINK-16921][e2e] Enable K8s tests
add 0c9ced4 [FLINK-17023][scripts] Improve handling of unexpected input
add b6956ff [FLINK-17132][metrics] Update licensing
add 935313b [hotfix][runtime] Fix import order in PipelinedRegionComputeUtil
add cc30a4e [FLINK-17181][runtime] Drop generic Types in Topology Interface
add f506f0d [FLINK-16683][scripts] Remove Windows scripts
add d6c782f [FLINK-16895][docs] Remove build_docs.bat
add f960a2e [FLINK-17142][orc] Bump ORC dependency version to 1.5.6
add bdc6693 [FLINK-16411][AZP] Use Maven cache in the same data center
add a20ed14 [FLINK-8864][sql-client] Add cli command history in sql client and make it avaible after restart
add 5915277 [FLINK-16823][table-planner-blink] Fix TIMESTAMPDIFF/TIMESTAMPADD yield incorrect results
add c1a21c6 [FLINK-15400][elasticsearch] Support dynamic index for Elasticsearch sink connector
add 957c49d [hotfix] Force not null types for literals with non null value
add e69328b [hotfix][table-api] Make ApiExpressionUtils#objectToExpressionConversion convert byte[] to a BINARY literal
add 6b3e037 [FLINK-16379][table] Introduce fromValues in TableEnvironment.
add 7a7aaec [FLINK-16600][k8s] Respect the rest.bind-port config option for the Kubernetes setup
add cb8ae28 [FLINK-17223][AZP] Clean up disk space on Azure-hosted builders
add 6e61339 [FLINK-17068][python][tests] Ensure the permission of scripts set correctly before executing the Python tests (#11805)
add e0d2db7 [FLINK-16366] [table] Introduce executeSql method in TableEnvironment
add c874700 [FLINK-17118][AZP] Use rmetzger/flink-ci:ubuntu-amd64-bcef226 image which contains gcc used for PyFlink cython tests
add 3cd8716 [FLINK-17118][python] Add Cython support for primitive data types
add 5456360 [FLINK-17119][python] Add Cython support for composite types (#11812)
add b8b0331 [hotfix][docs] Fix overlapping text
new c4ffce9 [hotfix] Fix the code style in BashJavaUtilsITCase
new 8d60c0e [FLINK-16874] Respect the dynamic options when calculating memory options in taskmanager.sh
new 6ee9c4a [hotfix][runtime] Code clean-ups in BashJavaUtils and its test cases.
new 317b6f5 [FLINK-16742][dist] config.sh does not assume BashJavaUtils always return two lines of results.
new b048e89 [FLINK-16742][runtime] Ignore unknown command line options for BashJavaUtils.
new e099019 [FLINK-16742][dist] Extend and use BashJavaUtils for JM memory configuration in start-up scripts.
new d3bc2a6 [FLINK-16472][dist] Remove unused memory configuration logics from config.sh.
new 0c43e44 [FLINK-16742][runtime] Refactor BashJavaUtils, move to separated package and extract configuration loading logics.
new 23f62bd [FLINK-16745][clients] Parse JobManagerProcessSpec from Configuration into ClusterSpecification
new 1a831ec [FLINK-16745][yarn] Start Yarn JM with FLIP-116 JVM memory args
new 73b0949 [FLINK-16745][k8s] Start Kubernetes JM with FLIP-116 JVM memory args
new 46e0dc3 [FLINK-16745][coordination] Remove unused container cutoff
new 0f1bbee [hotfix] Remove 'TaskManager' from MemoryBackwardsCompatibilityUtils#getLegacyTaskManagerHeapMemoryIfExplicitlyConfigured
new 34d1295 [FLINK-16746] Add default 'jobmanager.memory.process.size: 1472m' to flink-conf.yaml
new 0c89116 [FLINK-16746][conf] Deprecate jobmanager.heap.size
new 5adb32b change type to cron
new 28b13e7 fixup! [FLINK-16745][k8s] Start Kubernetes JM with FLIP-116 JVM memory args
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (d433ea1c)
\
N -- N -- N refs/heads/flip116 (28b13e7)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 17 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
azure-pipelines.yml | 2 +-
docs/_config.yml | 1 -
docs/_config_dev_en.yml | 1 -
docs/_config_dev_zh.yml | 1 -
docs/build_docs.bat | 57 --
docs/dev/table/connect.md | 33 +-
docs/dev/table/connect.zh.md | 33 +-
docs/fig/program_dataflow.svg | 1062 ++++++++++----------
docs/monitoring/metrics.md | 8 +-
docs/monitoring/metrics.zh.md | 8 +-
docs/ops/deployment/local.md | 20 +-
docs/ops/deployment/local.zh.md | 20 +-
.../elasticsearch/ElasticsearchSinkBase.java | 2 +-
.../elasticsearch/ElasticsearchSinkFunction.java | 5 +
.../ElasticsearchUpsertTableSinkBase.java | 35 +-
.../index/AbstractTimeIndexGenerator.java | 26 +-
.../elasticsearch/index/IndexGenerator.java | 29 +-
.../elasticsearch/index/IndexGeneratorBase.java | 35 +-
.../elasticsearch/index/IndexGeneratorFactory.java | 279 +++++
.../elasticsearch/index/StaticIndexGenerator.java | 18 +-
...lasticsearchUpsertTableSinkFactoryTestBase.java | 11 +-
.../elasticsearch/index/IndexGeneratorTest.java | 246 +++++
.../Elasticsearch6UpsertTableSinkFactoryTest.java | 9 +-
.../Elasticsearch7UpsertTableSinkFactoryTest.java | 27 +-
flink-connectors/flink-connector-hive/pom.xml | 8 +-
.../table/catalog/hive/util/HiveTableUtilTest.java | 8 +-
.../flink-sql-connector-hive-1.2.2/pom.xml | 2 +-
.../flink-sql-connector-hive-2.2.0/pom.xml | 2 +-
flink-dist/src/main/flink-bin/bin/flink.bat | 33 -
.../src/main/flink-bin/bin/start-cluster.bat | 78 --
flink-end-to-end-tests/run-nightly-tests.sh | 5 +-
flink-end-to-end-tests/test-scripts/common.sh | 4 +-
.../test-scripts/common_kubernetes.sh | 25 +
.../test-scripts/test_kubernetes_session.sh | 12 +-
.../test_resume_externalized_checkpoints.sh | 2 -
.../test-scripts/test_resume_savepoint.sh | 2 -
.../test-scripts/test_table_shaded_dependencies.sh | 2 +
flink-formats/flink-orc-nohive/pom.xml | 4 +
.../orc/nohive/vector/AbstractOrcNoHiveVector.java | 8 +-
flink-formats/flink-orc/pom.xml | 4 +
.../flink/api/java/ExecutionEnvironment.java | 90 +-
.../apache/flink/api/java/utils/PlanGenerator.java | 191 ++++
.../flink/api/java/utils/PlanGeneratorTest.java | 77 ++
.../kubernetes/KubernetesClusterDescriptor.java | 6 +-
.../decorators/ExternalServiceDecorator.java | 1 +
.../parameters/KubernetesJobManagerParameters.java | 4 +
.../apache/flink/kubernetes/utils/Constants.java | 1 +
.../kubeclient/KubernetesJobManagerTestBase.java | 2 +
.../decorators/ExternalServiceDecoratorTest.java | 1 +
.../flink/metrics/datadog/DatadogHttpReporter.java | 2 +
.../datadog/DatadogHttpReporterFactory.java | 22 +-
...he.flink.metrics.reporter.MetricReporterFactory | 16 +
.../src/main/resources/META-INF/NOTICE | 8 +-
.../apache/flink/metrics/slf4j/Slf4jReporter.java | 2 +
.../flink/metrics/slf4j/Slf4jReporterFactory.java | 35 +
...he.flink.metrics.reporter.MetricReporterFactory | 16 +
flink-python/dev/lint-python.sh | 4 +
flink-python/dev/pip_test_code.py | 0
.../pyflink/fn_execution/fast_coder_impl.pxd | 25 +
.../pyflink/fn_execution/fast_coder_impl.pyx | 246 +++++
.../pyflink/fn_execution/tests/test_fast_coders.py | 73 ++
flink-python/pyflink/table/table_environment.py | 14 +
.../table/tests/test_environment_completeness.py | 1 +
flink-python/tox.ini | 2 +-
.../runtime/checkpoint/CheckpointCoordinator.java | 733 +++++++-------
.../checkpoint/CompletedCheckpointStore.java | 4 -
.../runtime/checkpoint/PendingCheckpoint.java | 325 +++---
.../StandaloneCompletedCheckpointStore.java | 51 +-
.../ZooKeeperCompletedCheckpointStore.java | 195 ++--
.../runtime/executiongraph/ExecutionGraph.java | 15 +-
.../failover/flip1/ExecutionFailureHandler.java | 4 +-
.../failover/flip1/FailoverRegion.java | 6 +-
.../failover/flip1/FailoverStrategy.java | 2 +-
.../failover/flip1/PipelinedRegionComputeUtil.java | 6 +-
.../failover/flip1/RestartAllFailoverStrategy.java | 6 +-
.../RestartPipelinedRegionFailoverStrategy.java | 24 +-
.../NotReleasingPartitionReleaseStrategy.java | 2 +-
.../partitionrelease/PartitionReleaseStrategy.java | 2 +-
.../RegionPartitionReleaseStrategy.java | 18 +-
.../flink/runtime/scheduler/SchedulerBase.java | 28 +-
.../adapter/DefaultExecutionTopology.java | 2 +-
.../scheduler/adapter/DefaultExecutionVertex.java | 2 +-
.../scheduler/adapter/DefaultResultPartition.java | 2 +-
.../strategy/EagerSchedulingStrategy.java | 6 +-
.../strategy/InputDependencyConstraintChecker.java | 32 +-
.../LazyFromSourcesSchedulingStrategy.java | 20 +-
.../strategy/SchedulingExecutionVertex.java | 4 +-
.../strategy/SchedulingPipelinedRegion.java | 2 +-
.../strategy/SchedulingResultPartition.java | 4 +-
.../strategy/SchedulingStrategyFactory.java | 2 +-
.../strategy/SchedulingStrategyUtils.java | 8 +-
.../scheduler/strategy/SchedulingTopology.java | 8 +-
.../flink/runtime/topology/BaseTopology.java | 2 +-
.../flink/runtime/topology/PipelinedRegion.java | 4 +-
.../org/apache/flink/runtime/topology/Result.java | 2 +-
.../apache/flink/runtime/topology/Topology.java | 2 +-
.../org/apache/flink/runtime/topology/Vertex.java | 4 +-
.../CheckpointCoordinatorFailureTest.java | 17 +-
.../CheckpointCoordinatorMasterHooksTest.java | 14 +-
.../CheckpointCoordinatorRestoringTest.java | 34 +-
.../checkpoint/CheckpointCoordinatorTest.java | 184 ++--
.../CheckpointCoordinatorTestingUtils.java | 16 +-
.../CheckpointCoordinatorTriggeringTest.java | 74 +-
.../checkpoint/CheckpointStateRestoreTest.java | 8 +-
.../ExecutionGraphCheckpointCoordinatorTest.java | 122 +--
.../FailoverStrategyCheckpointCoordinatorTest.java | 26 +-
.../runtime/checkpoint/PendingCheckpointTest.java | 212 +---
.../ZooKeeperCompletedCheckpointStoreITCase.java | 2 -
...egionStrategyNGAbortPendingCheckpointsTest.java | 10 +-
.../executiongraph/ArchivedExecutionGraphTest.java | 4 +-
.../flip1/ExecutionFailureHandlerTest.java | 2 +-
.../runtime/scheduler/DefaultSchedulerTest.java | 53 +-
.../InputDependencyConstraintCheckerTest.java | 2 +-
.../scheduler/strategy/TestSchedulingStrategy.java | 8 +-
.../strategy/TestingSchedulingExecutionVertex.java | 3 +-
.../strategy/TestingSchedulingResultPartition.java | 3 +-
.../strategy/TestingSchedulingTopology.java | 3 +-
flink-table/flink-sql-client/pom.xml | 8 -
.../org/apache/flink/table/client/SqlClient.java | 12 +-
.../table/client/cli/CliChangelogResultView.java | 4 +-
.../apache/flink/table/client/cli/CliClient.java | 18 +-
.../apache/flink/table/client/cli/CliOptions.java | 9 +-
.../flink/table/client/cli/CliOptionsParser.java | 20 +-
.../flink/table/client/cli/CliTableResultView.java | 3 +-
.../table/client/cli/CliTableauResultView.java | 119 +--
.../apache/flink/table/client/cli/CliUtils.java | 55 +-
.../src/main/resources/META-INF/NOTICE | 1 -
.../flink/table/client/cli/CliClientTest.java | 50 +-
.../flink/table/client/cli/CliResultViewTest.java | 7 +-
.../table/client/cli/CliTableauResultViewTest.java | 2 +-
.../flink/table/client/cli/CliUtilsTest.java | 68 --
flink-table/flink-sql-parser/pom.xml | 11 +-
.../org/apache/flink/table/api/ResultKind.java | 21 +-
.../apache/flink/table/api/TableEnvironment.java | 245 +++++
.../org/apache/flink/table/api/TableResult.java | 60 ++
.../table/api/internal/TableEnvironmentImpl.java | 193 +++-
.../flink/table/api/internal/TableResultImpl.java | 170 ++++
.../table/expressions/ApiExpressionUtils.java | 14 +-
.../expressions/resolver/ExpressionResolver.java | 33 +
.../table/operations/QueryOperationVisitor.java | 2 +
.../table/operations/ShowCatalogsOperation.java | 15 +-
.../table/operations/ShowDatabasesOperation.java | 15 +-
.../table/operations/ShowFunctionsOperation.java | 15 +-
.../flink/table/operations/ShowOperation.java | 14 +-
.../table/operations/ShowTablesOperation.java | 15 +-
.../table/operations/ValuesQueryOperation.java | 98 ++
.../operations/utils/OperationTreeBuilder.java | 50 +-
.../utils/QueryOperationDefaultVisitor.java | 6 +
.../operations/utils/ValuesOperationFactory.java | 373 +++++++
.../table/expressions/ObjectToExpressionTest.java | 14 +-
.../utils/ValuesOperationTreeBuilderTest.java | 435 ++++++++
flink-table/flink-table-api-scala/pom.xml | 7 +
.../expressions/ObjectToExpressionScalaTest.scala | 56 ++
flink-table/flink-table-common/pom.xml | 9 +
.../table/expressions/ValueLiteralExpression.java | 10 +-
.../org/apache/flink/table/utils/PrintUtils.java | 216 ++++
.../flink/table/expressions/ExpressionTest.java | 20 +-
.../apache/flink/table/utils/PrintUtilsTest.java | 201 ++++
flink-table/flink-table-planner-blink/pom.xml | 10 +
.../planner/expressions/ExpressionBuilder.java | 6 +-
.../functions/aggfunctions/AvgAggFunction.java | 4 +-
.../functions/aggfunctions/Count1AggFunction.java | 2 +-
.../functions/aggfunctions/CountAggFunction.java | 2 +-
.../functions/aggfunctions/ListAggFunction.java | 4 +-
.../aggfunctions/RankLikeAggFunctionBase.java | 2 +-
.../aggfunctions/SingleValueAggFunction.java | 4 +-
.../functions/aggfunctions/Sum0AggFunction.java | 14 +-
.../operations/SqlToOperationConverter.java | 36 +
.../planner/plan/QueryOperationConverter.java | 60 ++
.../src/main/resources/META-INF/NOTICE | 1 +
.../main/resources/META-INF/licenses/LICENSE.icu4j | 0
.../table/planner/calcite/FlinkPlannerImpl.scala | 7 +-
.../planner/codegen/calls/BuiltInMethods.scala | 8 +
.../planner/codegen/calls/ScalarOperatorGens.scala | 16 +-
.../codegen/calls/TimestampDiffCallGen.scala | 8 +-
.../planner/plan/utils/RexNodeExtractor.scala | 3 +-
.../planner/plan/stream/table/ValuesTest.java | 235 +++++
.../planner/runtime/stream/table/ValuesITCase.java | 447 ++++++++
.../table/planner/plan/stream/table/ValuesTest.xml | 263 +++++
.../flink/table/api/TableEnvironmentTest.scala | 183 +++-
.../planner/expressions/ScalarFunctionsTest.scala | 6 +-
.../planner/expressions/TemporalTypesTest.scala | 16 +
.../utils/TestCollectionTableFactory.scala | 5 +-
flink-table/flink-table-planner/pom.xml | 10 +
.../flink/table/plan/QueryOperationConverter.java | 60 ++
.../table/sqlexec/SqlToOperationConverter.java | 36 +
.../src/main/resources/META-INF/NOTICE | 1 +
.../main/resources/META-INF/licenses/LICENSE.icu4j | 0
.../flink/table/api/internal/TableEnvImpl.scala | 160 ++-
.../flink/table/calcite/FlinkPlannerImpl.scala | 7 +-
.../expressions/PlannerExpressionConverter.scala | 24 +-
.../expressions/PlannerExpressionParserImpl.scala | 2 +-
.../apache/flink/table/expressions/literals.scala | 3 +-
.../table/runtime/stream/table/ValuesITCase.java | 88 ++
.../api/batch/BatchTableEnvironmentTest.scala | 202 +++-
.../flink/table/utils/MockTableEnvironment.scala | 19 +-
.../table/runtime/functions/SqlFunctionUtils.java | 87 ++
flink-table/flink-table-uber-blink/pom.xml | 7 +
flink-table/flink-table-uber/pom.xml | 7 +
.../ResumeCheckpointManuallyITCase.java | 28 +-
pom.xml | 3 +-
tools/azure-pipelines/build-apache-repo.yml | 2 +-
tools/azure-pipelines/jobs-template.yml | 13 +
tools/ci/alibaba-mirror-settings.xml | 28 +
tools/ci/maven-utils.sh | 64 +-
205 files changed, 7394 insertions(+), 2823 deletions(-)
delete mode 100644 docs/build_docs.bat
copy flink-runtime/src/main/java/org/apache/flink/runtime/topology/Result.java => flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/index/AbstractTimeIndexGenerator.java (58%)
copy flink-runtime/src/main/java/org/apache/flink/runtime/topology/Result.java => flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/index/IndexGenerator.java (59%)
copy flink-runtime/src/main/java/org/apache/flink/runtime/topology/Result.java => flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/index/IndexGeneratorBase.java (55%)
create mode 100644 flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/index/IndexGeneratorFactory.java
copy flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/SchedulingStrategyFactory.java => flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/index/StaticIndexGenerator.java (69%)
create mode 100644 flink-connectors/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/index/IndexGeneratorTest.java
delete mode 100644 flink-dist/src/main/flink-bin/bin/flink.bat
delete mode 100644 flink-dist/src/main/flink-bin/bin/start-cluster.bat
create mode 100644 flink-java/src/main/java/org/apache/flink/api/java/utils/PlanGenerator.java
create mode 100644 flink-java/src/test/java/org/apache/flink/api/java/utils/PlanGeneratorTest.java
copy flink-runtime/src/main/java/org/apache/flink/runtime/topology/Vertex.java => flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpReporterFactory.java (64%)
create mode 100644 flink-metrics/flink-metrics-datadog/src/main/resources/META-INF/services/org.apache.flink.metrics.reporter.MetricReporterFactory
create mode 100644 flink-metrics/flink-metrics-slf4j/src/main/java/org/apache/flink/metrics/slf4j/Slf4jReporterFactory.java
create mode 100644 flink-metrics/flink-metrics-slf4j/src/main/resources/META-INF/services/org.apache.flink.metrics.reporter.MetricReporterFactory
mode change 100644 => 100755 flink-python/dev/pip_test_code.py
delete mode 100644 flink-table/flink-sql-client/src/test/java/org/apache/flink/table/client/cli/CliUtilsTest.java
copy flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/SchedulingStrategyFactory.java => flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/ResultKind.java (68%)
create mode 100644 flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableResult.java
create mode 100644 flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableResultImpl.java
copy flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/SchedulingStrategyFactory.java => flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ShowCatalogsOperation.java (70%)
copy flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/SchedulingStrategyFactory.java => flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ShowDatabasesOperation.java (70%)
copy flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/SchedulingStrategyFactory.java => flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ShowFunctionsOperation.java (70%)
copy flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/SchedulingStrategyFactory.java => flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ShowOperation.java (70%)
copy flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/SchedulingStrategyFactory.java => flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ShowTablesOperation.java (70%)
create mode 100644 flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/ValuesQueryOperation.java
create mode 100644 flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/ValuesOperationFactory.java
create mode 100644 flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/operations/utils/ValuesOperationTreeBuilderTest.java
create mode 100644 flink-table/flink-table-api-scala/src/test/scala/org/apache/flink/table/expressions/ObjectToExpressionScalaTest.scala
create mode 100644 flink-table/flink-table-common/src/main/java/org/apache/flink/table/utils/PrintUtils.java
create mode 100644 flink-table/flink-table-common/src/test/java/org/apache/flink/table/utils/PrintUtilsTest.java
copy flink-table/{flink-sql-client => flink-table-planner-blink}/src/main/resources/META-INF/licenses/LICENSE.icu4j (100%)
create mode 100644 flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/plan/stream/table/ValuesTest.java
create mode 100644 flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/stream/table/ValuesITCase.java
create mode 100644 flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/table/ValuesTest.xml
rename flink-table/{flink-sql-client => flink-table-planner}/src/main/resources/META-INF/licenses/LICENSE.icu4j (100%)
create mode 100644 flink-table/flink-table-planner/src/test/java/org/apache/flink/table/runtime/stream/table/ValuesITCase.java
create mode 100644 tools/ci/alibaba-mirror-settings.xml
[flink] 03/17: [hotfix][runtime] Code clean-ups in BashJavaUtils
and its test cases.
Posted by az...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
azagrebin pushed a commit to branch flip116
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 6ee9c4a6fc3b5dc7185d6b5a3169b4e62a8c0ab2
Author: Xintong Song <to...@gmail.com>
AuthorDate: Thu Mar 26 13:00:53 2020 +0800
[hotfix][runtime] Code clean-ups in BashJavaUtils and its test cases.
---
flink-dist/src/test/bin/runBashJavaUtilsCmd.sh | 2 +-
.../src/main/java/org/apache/flink/runtime/util/BashJavaUtils.java | 4 ++--
2 files changed, 3 insertions(+), 3 deletions(-)
diff --git a/flink-dist/src/test/bin/runBashJavaUtilsCmd.sh b/flink-dist/src/test/bin/runBashJavaUtilsCmd.sh
index e7b79ec..475d463 100755
--- a/flink-dist/src/test/bin/runBashJavaUtilsCmd.sh
+++ b/flink-dist/src/test/bin/runBashJavaUtilsCmd.sh
@@ -38,6 +38,6 @@ FLINK_DIST_JAR=`find $FLINK_TARGET_DIR -name 'flink-dist*.jar'`
. ${bin}/../../main/flink-bin/bin/config.sh > /dev/null
-output=$(runBashJavaUtilsCmd GET_TM_RESOURCE_PARAMS ${FLINK_CONF_DIR} "$FLINK_TARGET_DIR/bash-java-utils.jar:$FLINK_DIST_JAR}" $DYNAMIC_OPTS | tail -n 2)
+output=$(runBashJavaUtilsCmd ${COMMAND} ${FLINK_CONF_DIR} "$FLINK_TARGET_DIR/bash-java-utils.jar:$FLINK_DIST_JAR}" $DYNAMIC_OPTS | tail -n 2)
extractExecutionParams "$(echo "$output" | head -n 1)"
extractExecutionParams "$(echo "$output" | tail -n 1)"
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/BashJavaUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/BashJavaUtils.java
index 27b3c50..14d24f5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/util/BashJavaUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/BashJavaUtils.java
@@ -41,7 +41,7 @@ public class BashJavaUtils {
switch (Command.valueOf(args[0])) {
case GET_TM_RESOURCE_PARAMS:
- getTmResourceParams(args);
+ getTmResourceParams(Arrays.copyOfRange(args, 1, args.length));
break;
default:
// unexpected, Command#valueOf should fail if a unknown command is passed in
@@ -61,7 +61,7 @@ public class BashJavaUtils {
}
private static Configuration getConfigurationForStandaloneTaskManagers(String[] args) throws Exception {
- Configuration configuration = TaskManagerRunner.loadConfiguration(Arrays.copyOfRange(args, 1, args.length));
+ Configuration configuration = TaskManagerRunner.loadConfiguration(args);
return TaskExecutorProcessUtils.getConfigurationMapLegacyTaskManagerHeapSizeToConfigOption(
configuration, TaskManagerOptions.TOTAL_FLINK_MEMORY);
}
[flink] 09/17: [FLINK-16745][clients] Parse JobManagerProcessSpec
from Configuration into ClusterSpecification
Posted by az...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
azagrebin pushed a commit to branch flip116
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 23f62bd0f203a160beb60b625c97db6ee49f7660
Author: Andrey Zagrebin <az...@apache.org>
AuthorDate: Wed Apr 8 17:05:35 2020 +0300
[FLINK-16745][clients] Parse JobManagerProcessSpec from Configuration into ClusterSpecification
---
.../AbstractContainerizedClusterClientFactory.java | 10 +++++++---
.../flink/configuration/ConfigurationUtils.java | 19 -------------------
.../runtime/jobmanager/JobManagerProcessUtils.java | 8 +++++++-
3 files changed, 14 insertions(+), 23 deletions(-)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/deployment/AbstractContainerizedClusterClientFactory.java b/flink-clients/src/main/java/org/apache/flink/client/deployment/AbstractContainerizedClusterClientFactory.java
index 46af058..f78cae7 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/deployment/AbstractContainerizedClusterClientFactory.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/deployment/AbstractContainerizedClusterClientFactory.java
@@ -20,9 +20,10 @@ package org.apache.flink.client.deployment;
import org.apache.flink.annotation.Internal;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.ConfigurationUtils;
+import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.clusterframework.TaskExecutorProcessUtils;
+import org.apache.flink.runtime.jobmanager.JobManagerProcessUtils;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -36,8 +37,11 @@ public abstract class AbstractContainerizedClusterClientFactory<ClusterID> imple
public ClusterSpecification getClusterSpecification(Configuration configuration) {
checkNotNull(configuration);
- final int jobManagerMemoryMB = ConfigurationUtils
- .getJobManagerHeapMemory(configuration)
+ final int jobManagerMemoryMB = JobManagerProcessUtils.processSpecFromConfig(
+ JobManagerProcessUtils.getConfigurationWithLegacyHeapSizeMappedToNewConfigOption(
+ configuration,
+ JobManagerOptions.TOTAL_PROCESS_MEMORY))
+ .getTotalProcessMemorySize()
.getMebiBytes();
final int taskManagerMemoryMB = TaskExecutorProcessUtils
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ConfigurationUtils.java b/flink-core/src/main/java/org/apache/flink/configuration/ConfigurationUtils.java
index 0408a37..4fbc512 100755
--- a/flink-core/src/main/java/org/apache/flink/configuration/ConfigurationUtils.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/ConfigurationUtils.java
@@ -42,25 +42,6 @@ public class ConfigurationUtils {
private static final String[] EMPTY = new String[0];
/**
- * Get job manager's heap memory. This method will check the new key
- * {@link JobManagerOptions#JOB_MANAGER_HEAP_MEMORY} and
- * the old key {@link JobManagerOptions#JOB_MANAGER_HEAP_MEMORY_MB} for backwards compatibility.
- *
- * @param configuration the configuration object
- * @return the memory size of job manager's heap memory.
- */
- public static MemorySize getJobManagerHeapMemory(Configuration configuration) {
- if (configuration.containsKey(JobManagerOptions.JOB_MANAGER_HEAP_MEMORY.key())) {
- return configuration.get(JobManagerOptions.JOB_MANAGER_HEAP_MEMORY);
- } else if (configuration.containsKey(JobManagerOptions.JOB_MANAGER_HEAP_MEMORY_MB.key())) {
- return MemorySize.ofMebiBytes(configuration.getInteger(JobManagerOptions.JOB_MANAGER_HEAP_MEMORY_MB));
- } else {
- //use default value
- return JobManagerOptions.JOB_MANAGER_HEAP_MEMORY.defaultValue();
- }
- }
-
- /**
* @return extracted {@link MetricOptions#SYSTEM_RESOURCE_METRICS_PROBING_INTERVAL} or {@code Optional.empty()} if
* {@link MetricOptions#SYSTEM_RESOURCE_METRICS} are disabled.
*/
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/JobManagerProcessUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/JobManagerProcessUtils.java
index 62f61bf..82c2490 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/JobManagerProcessUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/JobManagerProcessUtils.java
@@ -24,8 +24,8 @@ import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.runtime.util.config.memory.CommonProcessMemorySpec;
import org.apache.flink.runtime.util.config.memory.JvmMetaspaceAndOverheadOptions;
-import org.apache.flink.runtime.util.config.memory.MemoryBackwardsCompatibilityUtils;
import org.apache.flink.runtime.util.config.memory.LegacyMemoryOptions;
+import org.apache.flink.runtime.util.config.memory.MemoryBackwardsCompatibilityUtils;
import org.apache.flink.runtime.util.config.memory.ProcessMemoryOptions;
import org.apache.flink.runtime.util.config.memory.ProcessMemoryUtils;
import org.apache.flink.runtime.util.config.memory.jobmanager.JobManagerFlinkMemory;
@@ -78,4 +78,10 @@ public class JobManagerProcessUtils {
ConfigOption<MemorySize> configOption) {
return LEGACY_MEMORY_UTILS.getConfWithLegacyHeapSizeMappedToNewConfigOption(configuration, configOption);
}
+
+ public static JobManagerProcessSpec createDefaultJobManagerProcessSpec(int totalProcessMemoryMb) {
+ Configuration configuration = new Configuration();
+ configuration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(totalProcessMemoryMb));
+ return processSpecFromConfig(configuration);
+ }
}
[flink] 13/17: [hotfix] Remove 'TaskManager' from
MemoryBackwardsCompatibilityUtils#getLegacyTaskManagerHeapMemoryIfExplicitlyConfigured
Posted by az...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
azagrebin pushed a commit to branch flip116
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 0f1bbee21ed6ccd0ace193ec27adf8e212ee35c5
Author: Andrey Zagrebin <az...@apache.org>
AuthorDate: Wed Apr 15 13:16:29 2020 +0300
[hotfix] Remove 'TaskManager' from MemoryBackwardsCompatibilityUtils#getLegacyTaskManagerHeapMemoryIfExplicitlyConfigured
---
.../runtime/util/config/memory/MemoryBackwardsCompatibilityUtils.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/config/memory/MemoryBackwardsCompatibilityUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/config/memory/MemoryBackwardsCompatibilityUtils.java
index f334922..80c7af3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/util/config/memory/MemoryBackwardsCompatibilityUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/config/memory/MemoryBackwardsCompatibilityUtils.java
@@ -49,7 +49,7 @@ public class MemoryBackwardsCompatibilityUtils {
if (configuration.contains(configOption)) {
return configuration;
}
- return getLegacyTaskManagerHeapMemoryIfExplicitlyConfigured(configuration)
+ return getLegacyHeapMemoryIfExplicitlyConfigured(configuration)
.map(legacyHeapSize -> {
Configuration copiedConfig = new Configuration(configuration);
copiedConfig.set(configOption, legacyHeapSize);
@@ -61,7 +61,7 @@ public class MemoryBackwardsCompatibilityUtils {
}).orElse(configuration);
}
- private Optional<MemorySize> getLegacyTaskManagerHeapMemoryIfExplicitlyConfigured(Configuration configuration) {
+ private Optional<MemorySize> getLegacyHeapMemoryIfExplicitlyConfigured(Configuration configuration) {
@SuppressWarnings("CallToSystemGetenv")
String totalProcessEnv = System.getenv(legacyMemoryOptions.getEnvVar());
if (totalProcessEnv != null) {
[flink] 01/17: [hotfix] Fix the code style in BashJavaUtilsITCase
Posted by az...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
azagrebin pushed a commit to branch flip116
in repository https://gitbox.apache.org/repos/asf/flink.git
commit c4ffce9d4485fc0f12797e15ab4c61583883d55c
Author: Yangze Guo <ka...@gmail.com>
AuthorDate: Thu Apr 16 17:11:16 2020 +0800
[hotfix] Fix the code style in BashJavaUtilsITCase
---
.../src/test/java/org/apache/flink/dist/BashJavaUtilsITCase.java | 9 +++++----
1 file changed, 5 insertions(+), 4 deletions(-)
diff --git a/flink-dist/src/test/java/org/apache/flink/dist/BashJavaUtilsITCase.java b/flink-dist/src/test/java/org/apache/flink/dist/BashJavaUtilsITCase.java
index 1281ed2..e0587f7 100644
--- a/flink-dist/src/test/java/org/apache/flink/dist/BashJavaUtilsITCase.java
+++ b/flink-dist/src/test/java/org/apache/flink/dist/BashJavaUtilsITCase.java
@@ -26,7 +26,8 @@ import org.junit.Test;
import java.util.Arrays;
import java.util.List;
-import static org.junit.Assert.assertEquals;
+import static org.hamcrest.core.Is.is;
+import static org.junit.Assert.assertThat;
/**
* Tests for BashJavaUtils.
@@ -42,8 +43,8 @@ public class BashJavaUtilsITCase extends JavaBashTestBase {
String[] commands = {RUN_BASH_JAVA_UTILS_CMD_SCRIPT, BashJavaUtils.Command.GET_TM_RESOURCE_PARAMS.toString()};
List<String> lines = Arrays.asList(executeScript(commands).split(System.lineSeparator()));
- assertEquals(2, lines.size());
- ConfigurationUtils.parseJvmArgString(lines.get(lines.size() - 2));
- ConfigurationUtils.parseTmResourceDynamicConfigs(lines.get(lines.size() - 1));
+ assertThat(lines.size(), is(2));
+ ConfigurationUtils.parseJvmArgString(lines.get(0));
+ ConfigurationUtils.parseTmResourceDynamicConfigs(lines.get(1));
}
}
[flink] 05/17: [FLINK-16742][runtime] Ignore unknown command line
options for BashJavaUtils.
Posted by az...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
azagrebin pushed a commit to branch flip116
in repository https://gitbox.apache.org/repos/asf/flink.git
commit b048e89f391a7f688be9539a542147f3c01bd7d6
Author: Xintong Song <to...@gmail.com>
AuthorDate: Thu Apr 9 13:09:58 2020 +0800
[FLINK-16742][runtime] Ignore unknown command line options for BashJavaUtils.
---
.../ClusterConfigurationParserFactory.java | 8 +-
.../runtime/taskexecutor/TaskManagerRunner.java | 21 +----
.../apache/flink/runtime/util/BashJavaUtils.java | 38 ++++++++-
.../runtime/util/ConfigurationParserUtils.java | 35 ++++++++
.../flink/runtime/util/BashJavaUtilsTest.java | 96 ++++++++++++++++++++++
5 files changed, 175 insertions(+), 23 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterConfigurationParserFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterConfigurationParserFactory.java
index 12115d7..069d5b13 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterConfigurationParserFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterConfigurationParserFactory.java
@@ -36,8 +36,7 @@ import static org.apache.flink.runtime.entrypoint.parser.CommandLineOptions.DYNA
*/
public class ClusterConfigurationParserFactory implements ParserResultFactory<ClusterConfiguration> {
- @Override
- public Options getOptions() {
+ public static Options options() {
final Options options = new Options();
options.addOption(CONFIG_DIR_OPTION);
options.addOption(DYNAMIC_PROPERTY_OPTION);
@@ -46,6 +45,11 @@ public class ClusterConfigurationParserFactory implements ParserResultFactory<Cl
}
@Override
+ public Options getOptions() {
+ return options();
+ }
+
+ @Override
public ClusterConfiguration createResult(@Nonnull CommandLine commandLine) {
final String configDir = commandLine.getOptionValue(CONFIG_DIR_OPTION.getOpt());
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
index 6fbd874..457b402 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
@@ -21,8 +21,6 @@ package org.apache.flink.runtime.taskexecutor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.ConfigurationUtils;
-import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.configuration.WebOptions;
import org.apache.flink.core.fs.FileSystem;
@@ -34,10 +32,7 @@ import org.apache.flink.runtime.blob.BlobCacheService;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.concurrent.FutureUtils;
import org.apache.flink.runtime.concurrent.ScheduledExecutor;
-import org.apache.flink.runtime.entrypoint.ClusterConfiguration;
-import org.apache.flink.runtime.entrypoint.ClusterConfigurationParserFactory;
import org.apache.flink.runtime.entrypoint.FlinkParseException;
-import org.apache.flink.runtime.entrypoint.parser.CommandLineParser;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServicesUtils;
@@ -55,6 +50,7 @@ import org.apache.flink.runtime.rpc.akka.AkkaRpcServiceUtils;
import org.apache.flink.runtime.security.SecurityConfiguration;
import org.apache.flink.runtime.security.SecurityUtils;
import org.apache.flink.runtime.taskmanager.MemoryLogger;
+import org.apache.flink.runtime.util.ConfigurationParserUtils;
import org.apache.flink.runtime.util.EnvironmentInformation;
import org.apache.flink.runtime.util.ExecutorThreadFactory;
import org.apache.flink.runtime.util.Hardware;
@@ -292,20 +288,7 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
}
public static Configuration loadConfiguration(String[] args) throws FlinkParseException {
- final CommandLineParser<ClusterConfiguration> commandLineParser = new CommandLineParser<>(new ClusterConfigurationParserFactory());
-
- final ClusterConfiguration clusterConfiguration;
-
- try {
- clusterConfiguration = commandLineParser.parse(args);
- } catch (FlinkParseException e) {
- LOG.error("Could not parse the command line options.", e);
- commandLineParser.printHelp(TaskManagerRunner.class.getSimpleName());
- throw e;
- }
-
- final Configuration dynamicProperties = ConfigurationUtils.createConfiguration(clusterConfiguration.getDynamicProperties());
- return GlobalConfiguration.loadConfiguration(clusterConfiguration.getConfigDir(), dynamicProperties);
+ return ConfigurationParserUtils.loadCommonConfiguration(args, TaskManagerRunner.class.getSimpleName());
}
public static void runTaskManager(Configuration configuration, ResourceID resourceId, PluginManager pluginManager) throws Exception {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/BashJavaUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/BashJavaUtils.java
index e00d667..1478e1c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/util/BashJavaUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/BashJavaUtils.java
@@ -23,10 +23,16 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.clusterframework.TaskExecutorProcessSpec;
import org.apache.flink.runtime.clusterframework.TaskExecutorProcessUtils;
-import org.apache.flink.runtime.taskexecutor.TaskManagerRunner;
+import org.apache.flink.runtime.entrypoint.ClusterConfigurationParserFactory;
import org.apache.flink.runtime.util.config.memory.ProcessMemoryUtils;
+import org.apache.flink.util.FlinkException;
+import org.apache.commons.cli.Options;
+
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
import static org.apache.flink.util.Preconditions.checkArgument;
@@ -63,11 +69,39 @@ public class BashJavaUtils {
}
private static Configuration getConfigurationForStandaloneTaskManagers(String[] args) throws Exception {
- Configuration configuration = TaskManagerRunner.loadConfiguration(args);
+ Configuration configuration = loadConfiguration(args);
return TaskExecutorProcessUtils.getConfigurationMapLegacyTaskManagerHeapSizeToConfigOption(
configuration, TaskManagerOptions.TOTAL_FLINK_MEMORY);
}
+ @VisibleForTesting
+ static Configuration loadConfiguration(String[] args) throws FlinkException {
+ return ConfigurationParserUtils.loadCommonConfiguration(
+ filterCmdArgs(args),
+ BashJavaUtils.class.getSimpleName());
+ }
+
+ private static String[] filterCmdArgs(String[] args) {
+ final Options options = ClusterConfigurationParserFactory.options();
+ final List<String> filteredArgs = new ArrayList<>();
+ final Iterator<String> iter = Arrays.asList(args).iterator();
+
+ while (iter.hasNext()) {
+ String token = iter.next();
+ if (options.hasOption(token)) {
+ filteredArgs.add(token);
+ if (options.getOption(token).hasArg() && iter.hasNext()) {
+ filteredArgs.add(iter.next());
+ }
+ } else if (token.startsWith("-D")) {
+ // "-Dkey=value"
+ filteredArgs.add(token);
+ }
+ }
+
+ return filteredArgs.toArray(new String[0]);
+ }
+
/**
* Commands that BashJavaUtils supports.
*/
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/ConfigurationParserUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/ConfigurationParserUtils.java
index 1104d83..ec0a7ff 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/util/ConfigurationParserUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/ConfigurationParserUtils.java
@@ -19,11 +19,20 @@
package org.apache.flink.runtime.util;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ConfigurationUtils;
+import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.configuration.TaskManagerOptions;
+import org.apache.flink.runtime.entrypoint.ClusterConfiguration;
+import org.apache.flink.runtime.entrypoint.ClusterConfigurationParserFactory;
+import org.apache.flink.runtime.entrypoint.FlinkParseException;
+import org.apache.flink.runtime.entrypoint.parser.CommandLineParser;
import org.apache.flink.runtime.memory.MemoryManager;
import org.apache.flink.util.MathUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import static org.apache.flink.util.MathUtils.checkedDownCast;
/**
@@ -32,6 +41,8 @@ import static org.apache.flink.util.MathUtils.checkedDownCast;
*/
public class ConfigurationParserUtils {
+ private static final Logger LOG = LoggerFactory.getLogger(ConfigurationParserUtils.class);
+
/**
* Parses the configuration to get the number of slots and validates the value.
*
@@ -95,4 +106,28 @@ public class ConfigurationParserUtils {
return pageSize;
}
+
+ /**
+ * Generate configuration from only the config file and dynamic properties.
+ * @param args the commandline arguments
+ * @param cmdLineSyntax the syntax for this application
+ * @return generated configuration
+ * @throws FlinkParseException if the configuration cannot be generated
+ */
+ public static Configuration loadCommonConfiguration(String[] args, String cmdLineSyntax) throws FlinkParseException {
+ final CommandLineParser<ClusterConfiguration> commandLineParser = new CommandLineParser<>(new ClusterConfigurationParserFactory());
+
+ final ClusterConfiguration clusterConfiguration;
+
+ try {
+ clusterConfiguration = commandLineParser.parse(args);
+ } catch (FlinkParseException e) {
+ LOG.error("Could not parse the command line options.", e);
+ commandLineParser.printHelp(cmdLineSyntax);
+ throw e;
+ }
+
+ final Configuration dynamicProperties = ConfigurationUtils.createConfiguration(clusterConfiguration.getDynamicProperties());
+ return GlobalConfiguration.loadConfiguration(clusterConfiguration.getConfigDir(), dynamicProperties);
+ }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/util/BashJavaUtilsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/util/BashJavaUtilsTest.java
new file mode 100644
index 0000000..413a6fe
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/util/BashJavaUtilsTest.java
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.util;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.io.FileWriter;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+/**
+ * Tests for {@link BashJavaUtils}.
+ */
+public class BashJavaUtilsTest extends TestLogger {
+
+ private static final String TEST_CONFIG_KEY = "test.key";
+ private static final String TEST_CONFIG_VALUE = "test_value";
+
+ @Rule
+ public TemporaryFolder confDir = new TemporaryFolder() {
+ @Override
+ protected void before() throws Throwable {
+ super.create();
+ File flinkConfFile = newFile("flink-conf.yaml");
+ FileWriter fw = new FileWriter(flinkConfFile);
+ fw.write(TEST_CONFIG_KEY + ": " + TEST_CONFIG_VALUE + "\n");
+ fw.close();
+ }
+ };
+
+ @Test
+ public void testLoadConfigurationConfigDirLongOpt() throws Exception {
+ String[] args = {"--configDir", confDir.getRoot().getAbsolutePath()};
+ Configuration configuration = BashJavaUtils.loadConfiguration(args);
+ verifyConfiguration(configuration, TEST_CONFIG_KEY, TEST_CONFIG_VALUE);
+ }
+
+ @Test
+ public void testLoadConfigurationConfigDirShortOpt() throws Exception {
+ String[] args = {"-c", confDir.getRoot().getAbsolutePath()};
+ Configuration configuration = BashJavaUtils.loadConfiguration(args);
+ verifyConfiguration(configuration, TEST_CONFIG_KEY, TEST_CONFIG_VALUE);
+ }
+
+ @Test
+ public void testLoadConfigurationDynamicPropertyWithSpace() throws Exception {
+ String[] args = {"--configDir", confDir.getRoot().getAbsolutePath(), "-D", "key=value"};
+ Configuration configuration = BashJavaUtils.loadConfiguration(args);
+ verifyConfiguration(configuration, "key", "value");
+ }
+
+ @Test
+ public void testLoadConfigurationDynamicPropertyWithoutSpace() throws Exception {
+ String[] args = {"--configDir", confDir.getRoot().getAbsolutePath(), "-Dkey=value"};
+ Configuration configuration = BashJavaUtils.loadConfiguration(args);
+ verifyConfiguration(configuration, "key", "value");
+ }
+
+ @Test
+ public void testLoadConfigurationIgnoreUnknownToken() throws Exception {
+ String [] args = {"unknown", "-u", "--configDir", confDir.getRoot().getAbsolutePath(), "--unknown", "-Dkey=value"};
+ Configuration configuration = BashJavaUtils.loadConfiguration(args);
+ verifyConfiguration(configuration, TEST_CONFIG_KEY, TEST_CONFIG_VALUE);
+ verifyConfiguration(configuration, "key", "value");
+ }
+
+ private void verifyConfiguration(Configuration config, String key, String expectedValue) {
+ ConfigOption<String> option = key(key).stringType().noDefaultValue();
+ assertThat(config.get(option), is(expectedValue));
+ }
+}
[flink] 17/17: fixup! [FLINK-16745][k8s] Start Kubernetes JM with
FLIP-116 JVM memory args
Posted by az...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
azagrebin pushed a commit to branch flip116
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 28b13e79edd7003d07e2c2b61bc6b979b8cdf6dc
Author: Andrey Zagrebin <az...@apache.org>
AuthorDate: Sat Apr 18 20:00:09 2020 +0300
fixup! [FLINK-16745][k8s] Start Kubernetes JM with FLIP-116 JVM memory args
---
flink-end-to-end-tests/test-scripts/test_kubernetes_session.sh | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/flink-end-to-end-tests/test-scripts/test_kubernetes_session.sh b/flink-end-to-end-tests/test-scripts/test_kubernetes_session.sh
index b45e1c3..6dd4083 100755
--- a/flink-end-to-end-tests/test-scripts/test_kubernetes_session.sh
+++ b/flink-end-to-end-tests/test-scripts/test_kubernetes_session.sh
@@ -64,7 +64,7 @@ mkdir -p "$(dirname $LOCAL_OUTPUT_PATH)"
# Set the memory and cpu smaller than default, so that the jobmanager and taskmanager pods could be allocated in minikube.
"$FLINK_DIR"/bin/kubernetes-session.sh -Dkubernetes.cluster-id=${CLUSTER_ID} \
-Dkubernetes.container.image=${FLINK_IMAGE_NAME} \
- -Djobmanager.memory.process.size=768m \
+ -Djobmanager.memory.process.size=1088m \
-Dkubernetes.jobmanager.cpu=0.5 \
-Dkubernetes.taskmanager.cpu=0.5 \
-Dkubernetes.container-start-command-template="%java% %classpath% %jvmmem% %jvmopts% %logging% %class% %args%" \
[flink] 08/17: [FLINK-16742][runtime] Refactor BashJavaUtils,
move to separated package and extract configuration loading logics.
Posted by az...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
azagrebin pushed a commit to branch flip116
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 0c43e44f2a660a349c22a8e331c3a096dff7926a
Author: Xintong Song <to...@gmail.com>
AuthorDate: Wed Apr 15 17:25:27 2020 +0800
[FLINK-16742][runtime] Refactor BashJavaUtils, move to separated package and extract configuration loading logics.
---
flink-dist/pom.xml | 4 +-
flink-dist/src/main/flink-bin/bin/config.sh | 2 +-
.../org/apache/flink/dist/BashJavaUtilsITCase.java | 2 +-
.../runtime/util/{ => bash}/BashJavaUtils.java | 41 +------------
.../flink/runtime/util/bash/FlinkConfigLoader.java | 68 ++++++++++++++++++++++
.../FlinkConfigLoaderTest.java} | 16 ++---
6 files changed, 83 insertions(+), 50 deletions(-)
diff --git a/flink-dist/pom.xml b/flink-dist/pom.xml
index a10b01d..20337cf 100644
--- a/flink-dist/pom.xml
+++ b/flink-dist/pom.xml
@@ -708,7 +708,7 @@ under the License.
<exclude>META-INF/*.SF</exclude>
<exclude>META-INF/*.DSA</exclude>
<exclude>META-INF/*.RSA</exclude>
- <exclude>org/apache/flink/runtime/util/BashJavaUtils.class</exclude>
+ <exclude>org/apache/flink/runtime/util/bash/BashJavaUtils.class</exclude>
</excludes>
</filter>
</filters>
@@ -769,7 +769,7 @@ under the License.
<filter>
<artifact>org.apache.flink:*</artifact>
<includes>
- <include>org/apache/flink/runtime/util/BashJavaUtils.class</include>
+ <include>org/apache/flink/runtime/util/bash/BashJavaUtils.class</include>
</includes>
</filter>
</filters>
diff --git a/flink-dist/src/main/flink-bin/bin/config.sh b/flink-dist/src/main/flink-bin/bin/config.sh
index a40a1d4..11ec69b 100755
--- a/flink-dist/src/main/flink-bin/bin/config.sh
+++ b/flink-dist/src/main/flink-bin/bin/config.sh
@@ -480,7 +480,7 @@ runBashJavaUtilsCmd() {
local dynamic_args=${@:4}
class_path=`manglePathList ${class_path}`
- local output=`${JAVA_RUN} -classpath ${class_path} org.apache.flink.runtime.util.BashJavaUtils ${cmd} --configDir ${conf_dir} $dynamic_args 2>&1 | tail -n 1000`
+ local output=`${JAVA_RUN} -classpath ${class_path} org.apache.flink.runtime.util.bash.BashJavaUtils ${cmd} --configDir ${conf_dir} $dynamic_args 2>&1 | tail -n 1000`
if [[ $? -ne 0 ]]; then
echo "[ERROR] Cannot run BashJavaUtils to execute command ${cmd}." 1>&2
# Print the output in case the user redirect the log to console.
diff --git a/flink-dist/src/test/java/org/apache/flink/dist/BashJavaUtilsITCase.java b/flink-dist/src/test/java/org/apache/flink/dist/BashJavaUtilsITCase.java
index 4604697..7ce5566 100644
--- a/flink-dist/src/test/java/org/apache/flink/dist/BashJavaUtilsITCase.java
+++ b/flink-dist/src/test/java/org/apache/flink/dist/BashJavaUtilsITCase.java
@@ -21,7 +21,7 @@ package org.apache.flink.dist;
import org.apache.flink.configuration.ConfigurationUtils;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.TaskManagerOptions;
-import org.apache.flink.runtime.util.BashJavaUtils;
+import org.apache.flink.runtime.util.bash.BashJavaUtils;
import org.junit.Test;
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/BashJavaUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/bash/BashJavaUtils.java
similarity index 76%
rename from flink-runtime/src/main/java/org/apache/flink/runtime/util/BashJavaUtils.java
rename to flink-runtime/src/main/java/org/apache/flink/runtime/util/bash/BashJavaUtils.java
index 666d09a..fb71938 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/util/BashJavaUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/bash/BashJavaUtils.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.flink.runtime.util;
+package org.apache.flink.runtime.util.bash;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.configuration.Configuration;
@@ -24,18 +24,11 @@ import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.clusterframework.TaskExecutorProcessSpec;
import org.apache.flink.runtime.clusterframework.TaskExecutorProcessUtils;
-import org.apache.flink.runtime.entrypoint.ClusterConfigurationParserFactory;
import org.apache.flink.runtime.jobmanager.JobManagerProcessSpec;
import org.apache.flink.runtime.jobmanager.JobManagerProcessUtils;
import org.apache.flink.runtime.util.config.memory.ProcessMemoryUtils;
-import org.apache.flink.util.FlinkException;
-import org.apache.commons.cli.Options;
-
-import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Iterator;
-import java.util.List;
import static org.apache.flink.util.Preconditions.checkArgument;
@@ -77,7 +70,7 @@ public class BashJavaUtils {
}
private static Configuration getConfigurationForStandaloneTaskManagers(String[] args) throws Exception {
- Configuration configuration = loadConfiguration(args);
+ Configuration configuration = FlinkConfigLoader.loadConfiguration(args);
return TaskExecutorProcessUtils.getConfigurationMapLegacyTaskManagerHeapSizeToConfigOption(
configuration, TaskManagerOptions.TOTAL_FLINK_MEMORY);
}
@@ -89,39 +82,11 @@ public class BashJavaUtils {
}
private static Configuration getConfigurationForStandaloneJobManager(String[] args) throws Exception {
- Configuration configuration = loadConfiguration(args);
+ Configuration configuration = FlinkConfigLoader.loadConfiguration(args);
return JobManagerProcessUtils.getConfigurationWithLegacyHeapSizeMappedToNewConfigOption(
configuration, JobManagerOptions.TOTAL_FLINK_MEMORY);
}
- @VisibleForTesting
- static Configuration loadConfiguration(String[] args) throws FlinkException {
- return ConfigurationParserUtils.loadCommonConfiguration(
- filterCmdArgs(args),
- BashJavaUtils.class.getSimpleName());
- }
-
- private static String[] filterCmdArgs(String[] args) {
- final Options options = ClusterConfigurationParserFactory.options();
- final List<String> filteredArgs = new ArrayList<>();
- final Iterator<String> iter = Arrays.asList(args).iterator();
-
- while (iter.hasNext()) {
- String token = iter.next();
- if (options.hasOption(token)) {
- filteredArgs.add(token);
- if (options.getOption(token).hasArg() && iter.hasNext()) {
- filteredArgs.add(iter.next());
- }
- } else if (token.startsWith("-D")) {
- // "-Dkey=value"
- filteredArgs.add(token);
- }
- }
-
- return filteredArgs.toArray(new String[0]);
- }
-
/**
* Commands that BashJavaUtils supports.
*/
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/bash/FlinkConfigLoader.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/bash/FlinkConfigLoader.java
new file mode 100644
index 0000000..36bbb48
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/bash/FlinkConfigLoader.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.util.bash;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.entrypoint.ClusterConfigurationParserFactory;
+import org.apache.flink.runtime.util.ConfigurationParserUtils;
+import org.apache.flink.util.FlinkException;
+
+import org.apache.commons.cli.Options;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+
+/**
+ * Util class for loading configuration from commandline arguments.
+ * It parses only the configuration file and dynamic properties, ignores other commandline options.
+ */
+public class FlinkConfigLoader {
+
+ private static final Options CMD_OPTIONS = ClusterConfigurationParserFactory.options();
+
+ public static Configuration loadConfiguration(String[] args) throws FlinkException {
+ return ConfigurationParserUtils.loadCommonConfiguration(
+ filterCmdArgs(args),
+ BashJavaUtils.class.getSimpleName());
+ }
+
+ private static String[] filterCmdArgs(String[] args) {
+ final List<String> filteredArgs = new ArrayList<>();
+ final Iterator<String> iter = Arrays.asList(args).iterator();
+
+ while (iter.hasNext()) {
+ String token = iter.next();
+ if (CMD_OPTIONS.hasOption(token)) {
+ filteredArgs.add(token);
+ if (CMD_OPTIONS.getOption(token).hasArg() && iter.hasNext()) {
+ filteredArgs.add(iter.next());
+ }
+ } else if (token.startsWith("-D")) {
+ // "-Dkey=value"
+ filteredArgs.add(token);
+ }
+ }
+
+ return filteredArgs.toArray(new String[0]);
+ }
+
+ private FlinkConfigLoader() {}
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/util/BashJavaUtilsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/util/bash/FlinkConfigLoaderTest.java
similarity index 86%
rename from flink-runtime/src/test/java/org/apache/flink/runtime/util/BashJavaUtilsTest.java
rename to flink-runtime/src/test/java/org/apache/flink/runtime/util/bash/FlinkConfigLoaderTest.java
index 413a6fe..c89ff82 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/util/BashJavaUtilsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/util/bash/FlinkConfigLoaderTest.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.flink.runtime.util;
+package org.apache.flink.runtime.util.bash;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.Configuration;
@@ -34,9 +34,9 @@ import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
/**
- * Tests for {@link BashJavaUtils}.
+ * Tests for {@link FlinkConfigLoader}.
*/
-public class BashJavaUtilsTest extends TestLogger {
+public class FlinkConfigLoaderTest extends TestLogger {
private static final String TEST_CONFIG_KEY = "test.key";
private static final String TEST_CONFIG_VALUE = "test_value";
@@ -56,35 +56,35 @@ public class BashJavaUtilsTest extends TestLogger {
@Test
public void testLoadConfigurationConfigDirLongOpt() throws Exception {
String[] args = {"--configDir", confDir.getRoot().getAbsolutePath()};
- Configuration configuration = BashJavaUtils.loadConfiguration(args);
+ Configuration configuration = FlinkConfigLoader.loadConfiguration(args);
verifyConfiguration(configuration, TEST_CONFIG_KEY, TEST_CONFIG_VALUE);
}
@Test
public void testLoadConfigurationConfigDirShortOpt() throws Exception {
String[] args = {"-c", confDir.getRoot().getAbsolutePath()};
- Configuration configuration = BashJavaUtils.loadConfiguration(args);
+ Configuration configuration = FlinkConfigLoader.loadConfiguration(args);
verifyConfiguration(configuration, TEST_CONFIG_KEY, TEST_CONFIG_VALUE);
}
@Test
public void testLoadConfigurationDynamicPropertyWithSpace() throws Exception {
String[] args = {"--configDir", confDir.getRoot().getAbsolutePath(), "-D", "key=value"};
- Configuration configuration = BashJavaUtils.loadConfiguration(args);
+ Configuration configuration = FlinkConfigLoader.loadConfiguration(args);
verifyConfiguration(configuration, "key", "value");
}
@Test
public void testLoadConfigurationDynamicPropertyWithoutSpace() throws Exception {
String[] args = {"--configDir", confDir.getRoot().getAbsolutePath(), "-Dkey=value"};
- Configuration configuration = BashJavaUtils.loadConfiguration(args);
+ Configuration configuration = FlinkConfigLoader.loadConfiguration(args);
verifyConfiguration(configuration, "key", "value");
}
@Test
public void testLoadConfigurationIgnoreUnknownToken() throws Exception {
String [] args = {"unknown", "-u", "--configDir", confDir.getRoot().getAbsolutePath(), "--unknown", "-Dkey=value"};
- Configuration configuration = BashJavaUtils.loadConfiguration(args);
+ Configuration configuration = FlinkConfigLoader.loadConfiguration(args);
verifyConfiguration(configuration, TEST_CONFIG_KEY, TEST_CONFIG_VALUE);
verifyConfiguration(configuration, "key", "value");
}
[flink] 14/17: [FLINK-16746] Add default
'jobmanager.memory.process.size: 1472m' to flink-conf.yaml
Posted by az...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
azagrebin pushed a commit to branch flip116
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 34d12952348fa1091f607934b0577b1d5a7019be
Author: Andrey Zagrebin <az...@apache.org>
AuthorDate: Wed Apr 15 11:33:25 2020 +0300
[FLINK-16746] Add default 'jobmanager.memory.process.size: 1472m' to flink-conf.yaml
---
flink-dist/src/main/resources/flink-conf.yaml | 6 ++++--
1 file changed, 4 insertions(+), 2 deletions(-)
diff --git a/flink-dist/src/main/resources/flink-conf.yaml b/flink-dist/src/main/resources/flink-conf.yaml
index f58c04b..601bbcc 100644
--- a/flink-dist/src/main/resources/flink-conf.yaml
+++ b/flink-dist/src/main/resources/flink-conf.yaml
@@ -37,9 +37,11 @@ jobmanager.rpc.address: localhost
jobmanager.rpc.port: 6123
-# The heap size for the JobManager JVM
+# The total process memory size for the JobManager.
+#
+# Note this accounts for all memory usage within the JobManager process, including JVM metaspace and other overhead.
-jobmanager.heap.size: 1024m
+jobmanager.memory.process.size: 1472m
# The total process memory size for the TaskManager.
[flink] 06/17: [FLINK-16742][dist] Extend and use BashJavaUtils for
JM memory configuration in start-up scripts.
Posted by az...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
azagrebin pushed a commit to branch flip116
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e099019f80ad4420a35565ef7015a89c363d0b14
Author: Xintong Song <to...@gmail.com>
AuthorDate: Thu Apr 9 13:22:49 2020 +0800
[FLINK-16742][dist] Extend and use BashJavaUtils for JM memory configuration in start-up scripts.
This closes #11545.
---
flink-dist/src/main/flink-bin/bin/jobmanager.sh | 31 ++++++++++-----------
.../src/main/flink-bin/bin/standalone-job.sh | 30 ++++++++++----------
.../org/apache/flink/dist/BashJavaUtilsITCase.java | 32 +++++++++++++++++++++-
.../apache/flink/runtime/util/BashJavaUtils.java | 29 ++++++++++++++++++--
4 files changed, 88 insertions(+), 34 deletions(-)
diff --git a/flink-dist/src/main/flink-bin/bin/jobmanager.sh b/flink-dist/src/main/flink-bin/bin/jobmanager.sh
index 9efb83e..7403dd8 100755
--- a/flink-dist/src/main/flink-bin/bin/jobmanager.sh
+++ b/flink-dist/src/main/flink-bin/bin/jobmanager.sh
@@ -37,26 +37,16 @@ bin=`cd "$bin"; pwd`
ENTRYPOINT=standalonesession
if [[ $STARTSTOP == "start" ]] || [[ $STARTSTOP == "start-foreground" ]]; then
- if [ ! -z "${FLINK_JM_HEAP_MB}" ] && [ "${FLINK_JM_HEAP}" == 0 ]; then
- echo "used deprecated key \`${KEY_JOBM_MEM_MB}\`, please replace with key \`${KEY_JOBM_MEM_SIZE}\`"
- else
- flink_jm_heap_bytes=$(parseBytes ${FLINK_JM_HEAP})
- FLINK_JM_HEAP_MB=$(getMebiBytes ${flink_jm_heap_bytes})
- fi
-
- if [[ ! ${FLINK_JM_HEAP_MB} =~ $IS_NUMBER ]] || [[ "${FLINK_JM_HEAP_MB}" -lt "0" ]]; then
- echo "[ERROR] Configured JobManager memory size is not a valid value. Please set '${KEY_JOBM_MEM_SIZE}' in ${FLINK_CONF_FILE}."
- exit 1
- fi
-
- if [ "${FLINK_JM_HEAP_MB}" -gt "0" ]; then
- export JVM_ARGS="$JVM_ARGS -Xms"$FLINK_JM_HEAP_MB"m -Xmx"$FLINK_JM_HEAP_MB"m"
- fi
-
# Add JobManager-specific JVM options
export FLINK_ENV_JAVA_OPTS="${FLINK_ENV_JAVA_OPTS} ${FLINK_ENV_JAVA_OPTS_JM}"
# Startup parameters
+
+ java_utils_output=$(runBashJavaUtilsCmd GET_JM_RESOURCE_PARAMS ${FLINK_CONF_DIR} $FLINK_BIN_DIR/bash-java-utils.jar:$(findFlinkDistJar))
+ logging_output=$(extractLoggingOutputs "${java_utils_output}")
+ jvm_params=$(extractExecutionResults "${java_utils_output}" 1)
+ export JVM_ARGS="${JVM_ARGS} ${jvm_params}"
+
args=("--configDir" "${FLINK_CONF_DIR}" "--executionMode" "cluster")
if [ ! -z $HOST ]; then
args+=("--host")
@@ -67,6 +57,15 @@ if [[ $STARTSTOP == "start" ]] || [[ $STARTSTOP == "start-foreground" ]]; then
args+=("--webui-port")
args+=("${WEBUIPORT}")
fi
+
+ export FLINK_INHERITED_LOGS="
+$FLINK_INHERITED_LOGS
+
+JM_RESOURCE_PARAMS extraction logs:
+jvm_params: $jvm_params
+logs: $logging_output
+"
+
fi
if [[ $STARTSTOP == "start-foreground" ]]; then
diff --git a/flink-dist/src/main/flink-bin/bin/standalone-job.sh b/flink-dist/src/main/flink-bin/bin/standalone-job.sh
index 8a98a19..4df02ee 100644
--- a/flink-dist/src/main/flink-bin/bin/standalone-job.sh
+++ b/flink-dist/src/main/flink-bin/bin/standalone-job.sh
@@ -37,24 +37,24 @@ bin=`cd "$bin"; pwd`
ARGS=("--configDir" "${FLINK_CONF_DIR}" "${@:2}")
if [[ $STARTSTOP == "start" ]] || [[ $STARTSTOP == "start-foreground" ]]; then
- if [ ! -z "${FLINK_JM_HEAP_MB}" ] && [ "${FLINK_JM_HEAP}" == 0 ]; then
- echo "used deprecated key \`${KEY_JOBM_MEM_MB}\`, please replace with key \`${KEY_JOBM_MEM_SIZE}\`"
- else
- flink_jm_heap_bytes=$(parseBytes ${FLINK_JM_HEAP})
- FLINK_JM_HEAP_MB=$(getMebiBytes ${flink_jm_heap_bytes})
- fi
+ # Add cluster entry point specific JVM options
+ export FLINK_ENV_JAVA_OPTS="${FLINK_ENV_JAVA_OPTS} ${FLINK_ENV_JAVA_OPTS_JM}"
- if [[ ! ${FLINK_JM_HEAP_MB} =~ $IS_NUMBER ]] || [[ "${FLINK_JM_HEAP_MB}" -lt "0" ]]; then
- echo "[ERROR] Configured memory size is not a valid value. Please set '${KEY_JOBM_MEM_SIZE}' in ${FLINK_CONF_FILE}."
- exit 1
- fi
+ # Startup parameters
- if [ "${FLINK_JM_HEAP_MB}" -gt "0" ]; then
- export JVM_ARGS="$JVM_ARGS -Xms"$FLINK_JM_HEAP_MB"m -Xmx"$FLINK_JM_HEAP_MB"m"
- fi
+ java_utils_output=$(runBashJavaUtilsCmd GET_JM_RESOURCE_PARAMS ${FLINK_CONF_DIR} $FLINK_BIN_DIR/bash-java-utils.jar:$(findFlinkDistJar) "${ARGS[@]}")
+ logging_output=$(extractLoggingOutputs "${java_utils_output}")
+ jvm_params=$(extractExecutionResults "${java_utils_output}" 1)
+ export JVM_ARGS="${JVM_ARGS} ${jvm_params}"
+
+ export FLINK_INHERITED_LOGS="
+$FLINK_INHERITED_LOGS
+
+JM_RESOURCE_PARAMS extraction logs:
+jvm_params: $jvm_params
+logs: $logging_output
+"
- # Add cluster entry point specific JVM options
- export FLINK_ENV_JAVA_OPTS="${FLINK_ENV_JAVA_OPTS} ${FLINK_ENV_JAVA_OPTS_JM}"
fi
if [[ $STARTSTOP == "start-foreground" ]]; then
diff --git a/flink-dist/src/test/java/org/apache/flink/dist/BashJavaUtilsITCase.java b/flink-dist/src/test/java/org/apache/flink/dist/BashJavaUtilsITCase.java
index 8aafb23..4604697 100644
--- a/flink-dist/src/test/java/org/apache/flink/dist/BashJavaUtilsITCase.java
+++ b/flink-dist/src/test/java/org/apache/flink/dist/BashJavaUtilsITCase.java
@@ -19,6 +19,7 @@
package org.apache.flink.dist;
import org.apache.flink.configuration.ConfigurationUtils;
+import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.util.BashJavaUtils;
@@ -54,7 +55,7 @@ public class BashJavaUtilsITCase extends JavaBashTestBase {
}
@Test
- public void testConfigOverwrittenByDynamicOpts() throws Exception {
+ public void testGetTmResourceParamsConfigsWithDynamicProperties() throws Exception {
int expectedResultLines = 2;
double cpuCores = 39.0;
String[] commands = {
@@ -70,6 +71,35 @@ public class BashJavaUtilsITCase extends JavaBashTestBase {
}
@Test
+ public void testGetJmResourceParams() throws Exception {
+ int expectedResultLines = 1;
+ String[] commands = {
+ RUN_BASH_JAVA_UTILS_CMD_SCRIPT,
+ BashJavaUtils.Command.GET_JM_RESOURCE_PARAMS.toString(),
+ String.valueOf(expectedResultLines)};
+ List<String> lines = Arrays.asList(executeScript(commands).split(System.lineSeparator()));
+
+ assertThat(lines.size(), is(expectedResultLines));
+ ConfigurationUtils.parseJvmArgString(lines.get(0));
+ }
+
+ @Test
+ public void testGetJmResourceParamsWithDynamicProperties() throws Exception {
+ int expectedResultLines = 1;
+ long metaspace = 123456789L;
+ String[] commands = {
+ RUN_BASH_JAVA_UTILS_CMD_SCRIPT,
+ BashJavaUtils.Command.GET_JM_RESOURCE_PARAMS.toString(),
+ String.valueOf(expectedResultLines),
+ "-D" + JobManagerOptions.JVM_METASPACE.key() + "=" + metaspace + "b"};
+ List<String> lines = Arrays.asList(executeScript(commands).split(System.lineSeparator()));
+
+ assertThat(lines.size(), is(expectedResultLines));
+ Map<String, String> params = ConfigurationUtils.parseJvmArgString(lines.get(0));
+ assertThat(Long.valueOf(params.get("-XX:MaxMetaspaceSize=")), is(metaspace));
+ }
+
+ @Test
public void testExtractLoggingOutputs() throws Exception {
StringBuilder input = new StringBuilder();
List<String> expectedOutput = new ArrayList<>();
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/BashJavaUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/BashJavaUtils.java
index 1478e1c..666d09a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/util/BashJavaUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/BashJavaUtils.java
@@ -20,10 +20,13 @@ package org.apache.flink.runtime.util;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.clusterframework.TaskExecutorProcessSpec;
import org.apache.flink.runtime.clusterframework.TaskExecutorProcessUtils;
import org.apache.flink.runtime.entrypoint.ClusterConfigurationParserFactory;
+import org.apache.flink.runtime.jobmanager.JobManagerProcessSpec;
+import org.apache.flink.runtime.jobmanager.JobManagerProcessUtils;
import org.apache.flink.runtime.util.config.memory.ProcessMemoryUtils;
import org.apache.flink.util.FlinkException;
@@ -47,9 +50,14 @@ public class BashJavaUtils {
public static void main(String[] args) throws Exception {
checkArgument(args.length > 0, "Command not specified.");
+ String[] commandArgs = Arrays.copyOfRange(args, 1, args.length);
+
switch (Command.valueOf(args[0])) {
case GET_TM_RESOURCE_PARAMS:
- getTmResourceParams(Arrays.copyOfRange(args, 1, args.length));
+ getTmResourceParams(commandArgs);
+ break;
+ case GET_JM_RESOURCE_PARAMS:
+ getJmResourceParams(commandArgs);
break;
default:
// unexpected, Command#valueOf should fail if a unknown command is passed in
@@ -74,6 +82,18 @@ public class BashJavaUtils {
configuration, TaskManagerOptions.TOTAL_FLINK_MEMORY);
}
+ private static void getJmResourceParams(String[] args) throws Exception {
+ Configuration configuration = getConfigurationForStandaloneJobManager(args);
+ JobManagerProcessSpec jobManagerProcessSpec = JobManagerProcessUtils.processSpecFromConfig(configuration);
+ System.out.println(EXECUTION_PREFIX + ProcessMemoryUtils.generateJvmParametersStr(jobManagerProcessSpec));
+ }
+
+ private static Configuration getConfigurationForStandaloneJobManager(String[] args) throws Exception {
+ Configuration configuration = loadConfiguration(args);
+ return JobManagerProcessUtils.getConfigurationWithLegacyHeapSizeMappedToNewConfigOption(
+ configuration, JobManagerOptions.TOTAL_FLINK_MEMORY);
+ }
+
@VisibleForTesting
static Configuration loadConfiguration(String[] args) throws FlinkException {
return ConfigurationParserUtils.loadCommonConfiguration(
@@ -109,6 +129,11 @@ public class BashJavaUtils {
/**
* Get JVM parameters and dynamic configs of task executor resources.
*/
- GET_TM_RESOURCE_PARAMS
+ GET_TM_RESOURCE_PARAMS,
+
+ /**
+ * Get JVM parameters and dynamic configs of job manager resources.
+ */
+ GET_JM_RESOURCE_PARAMS
}
}
[flink] 15/17: [FLINK-16746][conf] Deprecate jobmanager.heap.size
Posted by az...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
azagrebin pushed a commit to branch flip116
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 0c8911612f0acf57e0708d10a6daabaa1222911e
Author: Andrey Zagrebin <az...@apache.org>
AuthorDate: Wed Apr 15 17:03:05 2020 +0300
[FLINK-16746][conf] Deprecate jobmanager.heap.size
---
.../generated/all_jobmanager_section.html | 6 ---
.../_includes/generated/common_memory_section.html | 48 ++++++++++++++++++++++
.../generated/job_manager_configuration.html | 46 ++++++++++++++++++++-
docs/ops/config.md | 6 ++-
docs/ops/config.zh.md | 6 ++-
docs/ops/deployment/cluster_setup.md | 4 +-
docs/ops/deployment/cluster_setup.zh.md | 4 +-
docs/ops/deployment/kubernetes.md | 2 +-
docs/ops/deployment/kubernetes.zh.md | 2 +-
docs/ops/deployment/mesos.md | 4 +-
docs/ops/deployment/mesos.zh.md | 4 +-
.../flink/configuration/JobManagerOptions.java | 16 +++-----
flink-end-to-end-tests/test-scripts/common.sh | 2 +-
flink-jepsen/src/jepsen/flink/db.clj | 2 +-
14 files changed, 117 insertions(+), 35 deletions(-)
diff --git a/docs/_includes/generated/all_jobmanager_section.html b/docs/_includes/generated/all_jobmanager_section.html
index 3369667..6ef552c 100644
--- a/docs/_includes/generated/all_jobmanager_section.html
+++ b/docs/_includes/generated/all_jobmanager_section.html
@@ -27,12 +27,6 @@
<td>This option specifies how the job computation recovers from task failures. Accepted values are:<ul><li>'full': Restarts all tasks to recover the job.</li><li>'region': Restarts all tasks that could be affected by the task failure. More details can be found <a href="../dev/task_failure_recovery.html#restart-pipelined-region-failover-strategy">here</a>.</li></ul></td>
</tr>
<tr>
- <td><h5>jobmanager.heap.size</h5></td>
- <td style="word-wrap: break-word;">1 gb</td>
- <td>MemorySize</td>
- <td>JVM heap size for the JobManager.</td>
- </tr>
- <tr>
<td><h5>jobmanager.rpc.address</h5></td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
diff --git a/docs/_includes/generated/common_memory_section.html b/docs/_includes/generated/common_memory_section.html
index 329916f..ce0878c 100644
--- a/docs/_includes/generated/common_memory_section.html
+++ b/docs/_includes/generated/common_memory_section.html
@@ -9,6 +9,54 @@
</thead>
<tbody>
<tr>
+ <td><h5>jobmanager.memory.flink.size</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>MemorySize</td>
+ <td>Total Flink Memory size for the JobManager. This includes all the memory that a JobManager consumes, except for JVM Metaspace and JVM Overhead. It consists of JVM Heap Memory and Off-heap Memory. See also 'jobmanager.memory.process.size' for total process memory size configuration.</td>
+ </tr>
+ <tr>
+ <td><h5>jobmanager.memory.heap.size</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>MemorySize</td>
+ <td>JVM Heap Memory size for JobManager. The minimum recommended JVM Heap size is 128.000mb (134217728 bytes).</td>
+ </tr>
+ <tr>
+ <td><h5>jobmanager.memory.jvm-metaspace.size</h5></td>
+ <td style="word-wrap: break-word;">256 mb</td>
+ <td>MemorySize</td>
+ <td>JVM Metaspace Size for the JobManager.</td>
+ </tr>
+ <tr>
+ <td><h5>jobmanager.memory.jvm-overhead.fraction</h5></td>
+ <td style="word-wrap: break-word;">0.1</td>
+ <td>Float</td>
+ <td>Fraction of Total Process Memory to be reserved for JVM Overhead. This is off-heap memory reserved for JVM overhead, such as thread stack space, compile cache, etc. This includes native memory but not direct memory, and will not be counted when Flink calculates JVM max direct memory size parameter. The size of JVM Overhead is derived to make up the configured fraction of the Total Process Memory. If the derived size is less or greater than the configured min or max size, [...]
+ </tr>
+ <tr>
+ <td><h5>jobmanager.memory.jvm-overhead.max</h5></td>
+ <td style="word-wrap: break-word;">1 gb</td>
+ <td>MemorySize</td>
+ <td>Max JVM Overhead size for the JobManager. This is off-heap memory reserved for JVM overhead, such as thread stack space, compile cache, etc. This includes native memory but not direct memory, and will not be counted when Flink calculates JVM max direct memory size parameter. The size of JVM Overhead is derived to make up the configured fraction of the Total Process Memory. If the derived size is less or greater than the configured min or max size, the min or max size will [...]
+ </tr>
+ <tr>
+ <td><h5>jobmanager.memory.jvm-overhead.min</h5></td>
+ <td style="word-wrap: break-word;">192 mb</td>
+ <td>MemorySize</td>
+ <td>Min JVM Overhead size for the JobManager. This is off-heap memory reserved for JVM overhead, such as thread stack space, compile cache, etc. This includes native memory but not direct memory, and will not be counted when Flink calculates JVM max direct memory size parameter. The size of JVM Overhead is derived to make up the configured fraction of the Total Process Memory. If the derived size is less or greater than the configured min or max size, the min or max size will [...]
+ </tr>
+ <tr>
+ <td><h5>jobmanager.memory.off-heap.size</h5></td>
+ <td style="word-wrap: break-word;">128 mb</td>
+ <td>MemorySize</td>
+ <td>Off-heap Memory size for JobManager. The JVM direct memory limit of the Job Manager process (-XX:MaxDirectMemorySize) will be set to this value. This option covers all off-heap memory usage including direct and native memory allocation.</td>
+ </tr>
+ <tr>
+ <td><h5>jobmanager.memory.process.size</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>MemorySize</td>
+ <td>Total Process Memory size for the JobManager. This includes all the memory that a JobManager JVM process consumes, consisting of Total Flink Memory, JVM Metaspace, and JVM Overhead. In containerized setups, this should be set to the container memory. See also 'jobmanager.memory.flink.size' for Total Flink Memory size configuration.</td>
+ </tr>
+ <tr>
<td><h5>taskmanager.memory.flink.size</h5></td>
<td style="word-wrap: break-word;">(none)</td>
<td>MemorySize</td>
diff --git a/docs/_includes/generated/job_manager_configuration.html b/docs/_includes/generated/job_manager_configuration.html
index a5c6c83..52cfa13 100644
--- a/docs/_includes/generated/job_manager_configuration.html
+++ b/docs/_includes/generated/job_manager_configuration.html
@@ -33,10 +33,52 @@
<td>This option specifies how the job computation recovers from task failures. Accepted values are:<ul><li>'full': Restarts all tasks to recover the job.</li><li>'region': Restarts all tasks that could be affected by the task failure. More details can be found <a href="../dev/task_failure_recovery.html#restart-pipelined-region-failover-strategy">here</a>.</li></ul></td>
</tr>
<tr>
- <td><h5>jobmanager.heap.size</h5></td>
+ <td><h5>jobmanager.memory.flink.size</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>MemorySize</td>
+ <td>Total Flink Memory size for the JobManager. This includes all the memory that a JobManager consumes, except for JVM Metaspace and JVM Overhead. It consists of JVM Heap Memory and Off-heap Memory. See also 'jobmanager.memory.process.size' for total process memory size configuration.</td>
+ </tr>
+ <tr>
+ <td><h5>jobmanager.memory.heap.size</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>MemorySize</td>
+ <td>JVM Heap Memory size for JobManager. The minimum recommended JVM Heap size is 128.000mb (134217728 bytes).</td>
+ </tr>
+ <tr>
+ <td><h5>jobmanager.memory.jvm-metaspace.size</h5></td>
+ <td style="word-wrap: break-word;">256 mb</td>
+ <td>MemorySize</td>
+ <td>JVM Metaspace Size for the JobManager.</td>
+ </tr>
+ <tr>
+ <td><h5>jobmanager.memory.jvm-overhead.fraction</h5></td>
+ <td style="word-wrap: break-word;">0.1</td>
+ <td>Float</td>
+ <td>Fraction of Total Process Memory to be reserved for JVM Overhead. This is off-heap memory reserved for JVM overhead, such as thread stack space, compile cache, etc. This includes native memory but not direct memory, and will not be counted when Flink calculates JVM max direct memory size parameter. The size of JVM Overhead is derived to make up the configured fraction of the Total Process Memory. If the derived size is less or greater than the configured min or max size, [...]
+ </tr>
+ <tr>
+ <td><h5>jobmanager.memory.jvm-overhead.max</h5></td>
<td style="word-wrap: break-word;">1 gb</td>
<td>MemorySize</td>
- <td>JVM heap size for the JobManager.</td>
+ <td>Max JVM Overhead size for the JobManager. This is off-heap memory reserved for JVM overhead, such as thread stack space, compile cache, etc. This includes native memory but not direct memory, and will not be counted when Flink calculates JVM max direct memory size parameter. The size of JVM Overhead is derived to make up the configured fraction of the Total Process Memory. If the derived size is less or greater than the configured min or max size, the min or max size will [...]
+ </tr>
+ <tr>
+ <td><h5>jobmanager.memory.jvm-overhead.min</h5></td>
+ <td style="word-wrap: break-word;">192 mb</td>
+ <td>MemorySize</td>
+ <td>Min JVM Overhead size for the JobManager. This is off-heap memory reserved for JVM overhead, such as thread stack space, compile cache, etc. This includes native memory but not direct memory, and will not be counted when Flink calculates JVM max direct memory size parameter. The size of JVM Overhead is derived to make up the configured fraction of the Total Process Memory. If the derived size is less or greater than the configured min or max size, the min or max size will [...]
+ </tr>
+ <tr>
+ <td><h5>jobmanager.memory.off-heap.size</h5></td>
+ <td style="word-wrap: break-word;">128 mb</td>
+ <td>MemorySize</td>
+ <td>Off-heap Memory size for JobManager. The JVM direct memory limit of the Job Manager process (-XX:MaxDirectMemorySize) will be set to this value. This option covers all off-heap memory usage including direct and native memory allocation.</td>
+ </tr>
+ <tr>
+ <td><h5>jobmanager.memory.process.size</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>MemorySize</td>
+ <td>Total Process Memory size for the JobManager. This includes all the memory that a JobManager JVM process consumes, consisting of Total Flink Memory, JVM Metaspace, and JVM Overhead. In containerized setups, this should be set to the container memory. See also 'jobmanager.memory.flink.size' for Total Flink Memory size configuration.</td>
</tr>
<tr>
<td><h5>jobmanager.rpc.address</h5></td>
diff --git a/docs/ops/config.md b/docs/ops/config.md
index 992a838..98fe2f0 100644
--- a/docs/ops/config.md
+++ b/docs/ops/config.md
@@ -51,8 +51,10 @@ If you use Flink with [Yarn]({{site.baseurl}}/ops/deployment/yarn_setup.html), [
The default memory sizes support simple streaming/batch applications, but are too low to yield good performance for more complex applications.
- - `jobmanager.heap.size`: Sets the size of the *Flink Master* (JobManager / ResourceManager / Dispatcher) JVM heap.
- - `taskmanager.memory.process.size`: Total size of the TaskManager process, including everything. Flink will subtract some memory for the JVM's own memory requirements (metaspace and others), and divide and configure the rest automatically between its components (network, managed memory, JVM Heap, etc.).
+ - `jobmanager.memory.process.size`: Total size of the *Flink Master* (JobManager / ResourceManager / Dispatcher) process.
+ - `taskmanager.memory.process.size`: Total size of the TaskManager process.
+
+The total sizes include everything. Flink will subtract some memory for the JVM's own memory requirements (metaspace and others), and divide and configure the rest automatically between its components (JVM Heap, Off-Heap, for Task Managers also network, managed memory etc.).
These value are configured as memory sizes, for example *1536m* or *2g*.
diff --git a/docs/ops/config.zh.md b/docs/ops/config.zh.md
index 23b7ce1..b2c71da 100644
--- a/docs/ops/config.zh.md
+++ b/docs/ops/config.zh.md
@@ -51,8 +51,10 @@ If you use Flink with [Yarn]({{site.baseurl}}/ops/deployment/yarn_setup.html), [
The default memory sizes support simple streaming/batch applications, but are too low to yield good performance for more complex applications.
- - `jobmanager.heap.size`: Sets the size of the *Flink Master* (JobManager / ResourceManager / Dispatcher) JVM heap.
- - `taskmanager.memory.process.size`: Total size of the TaskManager process, including everything. Flink will subtract some memory for the JVM's own memory requirements (metaspace and others), and divide and configure the rest automatically between its components (network, managed memory, JVM Heap, etc.).
+ - `jobmanager.memory.process.size`: Total size of the *Flink Master* (JobManager / ResourceManager / Dispatcher) process.
+ - `taskmanager.memory.process.size`: Total size of the TaskManager process.
+
+The total sizes include everything. Flink will subtract some memory for the JVM's own memory requirements (metaspace and others), and divide and configure the rest automatically between its components (JVM Heap, Off-Heap, for Task Managers also network, managed memory etc.).
These value are configured as memory sizes, for example *1536m* or *2g*.
diff --git a/docs/ops/deployment/cluster_setup.md b/docs/ops/deployment/cluster_setup.md
index 5efd38d..75455d2 100644
--- a/docs/ops/deployment/cluster_setup.md
+++ b/docs/ops/deployment/cluster_setup.md
@@ -68,7 +68,7 @@ cd flink-*
After having extracted the system files, you need to configure Flink for the cluster by editing *conf/flink-conf.yaml*.
-Set the `jobmanager.rpc.address` key to point to your master node. You should also define the maximum amount of main memory Flink is allowed to allocate on each node by setting the `jobmanager.heap.size` and `taskmanager.memory.process.size` keys.
+Set the `jobmanager.rpc.address` key to point to your master node. You should also define the maximum amount of main memory Flink is allowed to allocate on each node by setting the `jobmanager.memory.process.size` and `taskmanager.memory.process.size` keys.
These values are given in MB. If some worker nodes have more main memory which you want to allocate to the Flink system you can overwrite the default value by setting `taskmanager.memory.process.size` or `taskmanager.memory.flink.size` in *conf/flink-conf.yaml* on those specific nodes.
@@ -106,7 +106,7 @@ Please see the [configuration page](../config.html) for details and additional c
In particular,
- * the amount of available memory per JobManager (`jobmanager.heap.size`),
+ * the amount of available memory per JobManager (`jobmanager.memory.process.size`),
* the amount of available memory per TaskManager (`taskmanager.memory.process.size` and check [memory setup guide](../memory/mem_tuning.html#configure-memory-for-standalone-deployment)),
* the number of available CPUs per machine (`taskmanager.numberOfTaskSlots`),
* the total number of CPUs in the cluster (`parallelism.default`) and
diff --git a/docs/ops/deployment/cluster_setup.zh.md b/docs/ops/deployment/cluster_setup.zh.md
index 55696f8..f9d6356 100644
--- a/docs/ops/deployment/cluster_setup.zh.md
+++ b/docs/ops/deployment/cluster_setup.zh.md
@@ -68,7 +68,7 @@ cd flink-*
After having extracted the system files, you need to configure Flink for the cluster by editing *conf/flink-conf.yaml*.
-Set the `jobmanager.rpc.address` key to point to your master node. You should also define the maximum amount of main memory Flink is allowed to allocate on each node by setting the `jobmanager.heap.size` and `taskmanager.memory.process.size` keys.
+Set the `jobmanager.rpc.address` key to point to your master node. You should also define the maximum amount of main memory Flink is allowed to allocate on each node by setting the `jobmanager.memory.process.size` and `taskmanager.memory.process.size` keys.
These values are given in MB. If some worker nodes have more main memory which you want to allocate to the Flink system you can overwrite the default value by setting setting `taskmanager.memory.process.size` or `taskmanager.memory.flink.size` in *conf/flink-conf.yaml* on those specific nodes.
@@ -106,7 +106,7 @@ Please see the [configuration page](../config.html) for details and additional c
In particular,
- * the amount of available memory per JobManager (`jobmanager.heap.size`),
+ * the amount of available memory per JobManager (`jobmanager.memory.process.size`),
* the amount of available memory per TaskManager (`taskmanager.memory.process.size` and check [memory setup guide](../memory/mem_tuning.html#configure-memory-for-standalone-deployment)),
* the number of available CPUs per machine (`taskmanager.numberOfTaskSlots`),
* the total number of CPUs in the cluster (`parallelism.default`) and
diff --git a/docs/ops/deployment/kubernetes.md b/docs/ops/deployment/kubernetes.md
index d798bff..adce518 100644
--- a/docs/ops/deployment/kubernetes.md
+++ b/docs/ops/deployment/kubernetes.md
@@ -143,7 +143,7 @@ data:
blob.server.port: 6124
jobmanager.rpc.port: 6123
taskmanager.rpc.port: 6122
- jobmanager.heap.size: 1024m
+ jobmanager.memory.process.size: 1472m
taskmanager.memory.process.size: 1024m
log4j.properties: |+
rootLogger.level = INFO
diff --git a/docs/ops/deployment/kubernetes.zh.md b/docs/ops/deployment/kubernetes.zh.md
index f5dd31a..d724075 100644
--- a/docs/ops/deployment/kubernetes.zh.md
+++ b/docs/ops/deployment/kubernetes.zh.md
@@ -143,7 +143,7 @@ data:
blob.server.port: 6124
jobmanager.rpc.port: 6123
taskmanager.rpc.port: 6122
- jobmanager.heap.size: 1024m
+ jobmanager.memory.process.size: 1472m
taskmanager.memory.process.size: 1024m
log4j.properties: |+
rootLogger.level = INFO
diff --git a/docs/ops/deployment/mesos.md b/docs/ops/deployment/mesos.md
index 1df1838..a273171 100644
--- a/docs/ops/deployment/mesos.md
+++ b/docs/ops/deployment/mesos.md
@@ -215,7 +215,7 @@ For example:
bin/mesos-appmaster.sh \
-Dmesos.master=master.foobar.org:5050 \
- -Djobmanager.heap.size=1024m \
+ -Djobmanager.memory.process.size=1472m \
-Djobmanager.rpc.port=6123 \
-Drest.port=8081 \
-Dtaskmanager.memory.process.size=3500m \
@@ -236,7 +236,7 @@ Here is an example configuration for Marathon:
{
"id": "flink",
- "cmd": "$FLINK_HOME/bin/mesos-appmaster.sh -Djobmanager.heap.size=1024m -Djobmanager.rpc.port=6123 -Drest.port=8081 -Dtaskmanager.memory.process.size=1024m -Dtaskmanager.numberOfTaskSlots=2 -Dparallelism.default=2 -Dmesos.resourcemanager.tasks.cpus=1",
+ "cmd": "$FLINK_HOME/bin/mesos-appmaster.sh -Djobmanager.memory.process.size=1472m -Djobmanager.rpc.port=6123 -Drest.port=8081 -Dtaskmanager.memory.process.size=1024m -Dtaskmanager.numberOfTaskSlots=2 -Dparallelism.default=2 -Dmesos.resourcemanager.tasks.cpus=1",
"cpus": 1.0,
"mem": 1024
}
diff --git a/docs/ops/deployment/mesos.zh.md b/docs/ops/deployment/mesos.zh.md
index 83fcd9a..2d08235 100644
--- a/docs/ops/deployment/mesos.zh.md
+++ b/docs/ops/deployment/mesos.zh.md
@@ -215,7 +215,7 @@ For example:
bin/mesos-appmaster.sh \
-Dmesos.master=master.foobar.org:5050 \
- -Djobmanager.heap.size=1024m \
+ -Djobmanager.memory.process.size=1472m \
-Djobmanager.rpc.port=6123 \
-Drest.port=8081 \
-Dtaskmanager.memory.process.size=3500m \
@@ -236,7 +236,7 @@ Here is an example configuration for Marathon:
{
"id": "flink",
- "cmd": "$FLINK_HOME/bin/mesos-appmaster.sh -Djobmanager.heap.size=1024m -Djobmanager.rpc.port=6123 -Drest.port=8081 -Dtaskmanager.memory.process.size=1024m -Dtaskmanager.numberOfTaskSlots=2 -Dparallelism.default=2 -Dmesos.resourcemanager.tasks.cpus=1",
+ "cmd": "$FLINK_HOME/bin/mesos-appmaster.sh -Djobmanager.memory.process.size=1472m -Djobmanager.rpc.port=6123 -Drest.port=8081 -Dtaskmanager.memory.process.size=1024m -Dtaskmanager.numberOfTaskSlots=2 -Dparallelism.default=2 -Dmesos.resourcemanager.tasks.cpus=1",
"cpus": 1.0,
"mem": 1024
}
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
index 8a54f1c..c0a5a7bb1 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/JobManagerOptions.java
@@ -102,30 +102,31 @@ public class JobManagerOptions {
/**
* JVM heap size for the JobManager with memory size.
+ * @deprecated use {@link #TOTAL_FLINK_MEMORY} for standalone setups and {@link #TOTAL_PROCESS_MEMORY} for containerized setups.
*/
+ @Deprecated
@Documentation.Section(Documentation.Sections.ALL_JOB_MANAGER)
public static final ConfigOption<MemorySize> JOB_MANAGER_HEAP_MEMORY =
key("jobmanager.heap.size")
.memoryType()
- .defaultValue(MemorySize.ofMebiBytes(1024))
+ .noDefaultValue()
.withDescription("JVM heap size for the JobManager.");
/**
* JVM heap size (in megabytes) for the JobManager.
- * @deprecated use {@link #JOB_MANAGER_HEAP_MEMORY}
+ * @deprecated use {@link #TOTAL_FLINK_MEMORY} for standalone setups and {@link #TOTAL_PROCESS_MEMORY} for containerized setups.
*/
@Deprecated
public static final ConfigOption<Integer> JOB_MANAGER_HEAP_MEMORY_MB =
key("jobmanager.heap.mb")
.intType()
- .defaultValue(1024)
+ .noDefaultValue()
.withDescription("JVM heap size (in megabytes) for the JobManager.");
/**
* Total Process Memory size for the JobManager.
*/
@Documentation.Section(Documentation.Sections.COMMON_MEMORY)
- @Documentation.ExcludeFromDocumentation("New JM memory model is still in development.")
public static final ConfigOption<MemorySize> TOTAL_PROCESS_MEMORY =
key("jobmanager.memory.process.size")
.memoryType()
@@ -139,7 +140,6 @@ public class JobManagerOptions {
* Total Flink Memory size for the JobManager.
*/
@Documentation.Section(Documentation.Sections.COMMON_MEMORY)
- @Documentation.ExcludeFromDocumentation("New JM memory model is still in development.")
public static final ConfigOption<MemorySize> TOTAL_FLINK_MEMORY =
key("jobmanager.memory.flink.size")
.memoryType()
@@ -154,7 +154,6 @@ public class JobManagerOptions {
* JVM Heap Memory size for the JobManager.
*/
@Documentation.Section(Documentation.Sections.COMMON_MEMORY)
- @Documentation.ExcludeFromDocumentation("New JM memory model is still in development.")
public static final ConfigOption<MemorySize> JVM_HEAP_MEMORY =
key("jobmanager.memory.heap.size")
.memoryType()
@@ -166,7 +165,6 @@ public class JobManagerOptions {
* Off-heap Memory size for the JobManager.
*/
@Documentation.Section(Documentation.Sections.COMMON_MEMORY)
- @Documentation.ExcludeFromDocumentation("New JM memory model is still in development.")
public static final ConfigOption<MemorySize> OFF_HEAP_MEMORY =
key("jobmanager.memory.off-heap.size")
.memoryType()
@@ -179,7 +177,6 @@ public class JobManagerOptions {
* JVM Metaspace Size for the JobManager.
*/
@Documentation.Section(Documentation.Sections.COMMON_MEMORY)
- @Documentation.ExcludeFromDocumentation("New JM memory model is still in development.")
public static final ConfigOption<MemorySize> JVM_METASPACE =
key("jobmanager.memory.jvm-metaspace.size")
.memoryType()
@@ -197,7 +194,6 @@ public class JobManagerOptions {
* Min JVM Overhead size for the JobManager.
*/
@Documentation.Section(Documentation.Sections.COMMON_MEMORY)
- @Documentation.ExcludeFromDocumentation("New JM memory model is still in development.")
public static final ConfigOption<MemorySize> JVM_OVERHEAD_MIN =
key("jobmanager.memory.jvm-overhead.min")
.memoryType()
@@ -208,7 +204,6 @@ public class JobManagerOptions {
* Max JVM Overhead size for the TaskExecutors.
*/
@Documentation.Section(Documentation.Sections.COMMON_MEMORY)
- @Documentation.ExcludeFromDocumentation("New JM memory model is still in development.")
public static final ConfigOption<MemorySize> JVM_OVERHEAD_MAX =
key("jobmanager.memory.jvm-overhead.max")
.memoryType()
@@ -219,7 +214,6 @@ public class JobManagerOptions {
* Fraction of Total Process Memory to be reserved for JVM Overhead.
*/
@Documentation.Section(Documentation.Sections.COMMON_MEMORY)
- @Documentation.ExcludeFromDocumentation("New JM memory model is still in development.")
public static final ConfigOption<Float> JVM_OVERHEAD_FRACTION =
key("jobmanager.memory.jvm-overhead.fraction")
.floatType()
diff --git a/flink-end-to-end-tests/test-scripts/common.sh b/flink-end-to-end-tests/test-scripts/common.sh
index 3aabb3d..8329c12 100644
--- a/flink-end-to-end-tests/test-scripts/common.sh
+++ b/flink-end-to-end-tests/test-scripts/common.sh
@@ -167,7 +167,7 @@ function create_ha_config() {
jobmanager.rpc.address: localhost
jobmanager.rpc.port: 6123
- jobmanager.heap.size: 1024m
+ jobmanager.memory.process.size: 1024m
taskmanager.memory.process.size: 1024m
taskmanager.numberOfTaskSlots: ${TASK_SLOTS_PER_TM_HA}
diff --git a/flink-jepsen/src/jepsen/flink/db.clj b/flink-jepsen/src/jepsen/flink/db.clj
index 24738c2..9d8a986 100644
--- a/flink-jepsen/src/jepsen/flink/db.clj
+++ b/flink-jepsen/src/jepsen/flink/db.clj
@@ -41,7 +41,7 @@
{:high-availability "zookeeper"
:high-availability.zookeeper.quorum (zookeeper-quorum test)
:high-availability.storageDir "hdfs:///flink/ha"
- :jobmanager.heap.size "2048m"
+ :jobmanager.memory.process.size "2496m"
:jobmanager.rpc.address node
:state.savepoints.dir "hdfs:///flink/savepoints"
:rest.address node
[flink] 11/17: [FLINK-16745][k8s] Start Kubernetes JM with FLIP-116
JVM memory args
Posted by az...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
azagrebin pushed a commit to branch flip116
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 73b09499ed2c4c8cbcc78ff02e32d943d0837e54
Author: Andrey Zagrebin <az...@apache.org>
AuthorDate: Wed Apr 8 18:31:12 2020 +0300
[FLINK-16745][k8s] Start Kubernetes JM with FLIP-116 JVM memory args
This closes #11675.
---
.../test-scripts/test_kubernetes_session.sh | 2 +-
.../decorators/JavaCmdJobManagerDecorator.java | 13 ++++++------
.../flink/kubernetes/KubernetesTestBase.java | 4 ++++
.../kubernetes/cli/KubernetesSessionCliTest.java | 23 +++++++++++-----------
.../kubeclient/KubernetesJobManagerTestBase.java | 2 ++
.../decorators/JavaCmdJobManagerDecoratorTest.java | 6 ++++--
.../KubernetesJobManagerParametersTest.java | 1 -
7 files changed, 30 insertions(+), 21 deletions(-)
diff --git a/flink-end-to-end-tests/test-scripts/test_kubernetes_session.sh b/flink-end-to-end-tests/test-scripts/test_kubernetes_session.sh
index 021027b..1e450ec 100755
--- a/flink-end-to-end-tests/test-scripts/test_kubernetes_session.sh
+++ b/flink-end-to-end-tests/test-scripts/test_kubernetes_session.sh
@@ -64,7 +64,7 @@ mkdir -p "$(dirname $LOCAL_OUTPUT_PATH)"
# Set the memory and cpu smaller than default, so that the jobmanager and taskmanager pods could be allocated in minikube.
"$FLINK_DIR"/bin/kubernetes-session.sh -Dkubernetes.cluster-id=${CLUSTER_ID} \
-Dkubernetes.container.image=${FLINK_IMAGE_NAME} \
- -Djobmanager.heap.size=512m \
+ -Djobmanager.memory.process.size=768m \
-Dcontainerized.heap-cutoff-min=100 \
-Dkubernetes.jobmanager.cpu=0.5 \
-Dkubernetes.taskmanager.cpu=0.5 \
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/JavaCmdJobManagerDecorator.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/JavaCmdJobManagerDecorator.java
index 11900f4..81b0541 100644
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/JavaCmdJobManagerDecorator.java
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/JavaCmdJobManagerDecorator.java
@@ -22,7 +22,9 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.kubeclient.FlinkPod;
import org.apache.flink.kubernetes.kubeclient.parameters.KubernetesJobManagerParameters;
import org.apache.flink.kubernetes.utils.KubernetesUtils;
-import org.apache.flink.runtime.clusterframework.BootstrapTools;
+import org.apache.flink.runtime.jobmanager.JobManagerProcessUtils;
+import org.apache.flink.runtime.util.config.memory.ProcessMemorySpec;
+import org.apache.flink.runtime.util.config.memory.ProcessMemoryUtils;
import io.fabric8.kubernetes.api.model.Container;
import io.fabric8.kubernetes.api.model.ContainerBuilder;
@@ -46,7 +48,7 @@ public class JavaCmdJobManagerDecorator extends AbstractKubernetesStepDecorator
public FlinkPod decorateFlinkPod(FlinkPod flinkPod) {
final String startCommand = getJobManagerStartCommand(
kubernetesJobManagerParameters.getFlinkConfiguration(),
- kubernetesJobManagerParameters.getJobManagerMemoryMB(),
+ JobManagerProcessUtils.processSpecFromConfig(kubernetesJobManagerParameters.getFlinkConfiguration()),
kubernetesJobManagerParameters.getFlinkConfDirInPod(),
kubernetesJobManagerParameters.getFlinkLogDirInPod(),
kubernetesJobManagerParameters.hasLogback(),
@@ -67,7 +69,7 @@ public class JavaCmdJobManagerDecorator extends AbstractKubernetesStepDecorator
* Generates the shell command to start a jobmanager for kubernetes.
*
* @param flinkConfig The Flink configuration.
- * @param jobManagerMemoryMb JobManager heap size.
+ * @param jobManagerProcessSpec JobManager process memory spec.
* @param configDirectory The configuration directory for the flink-conf.yaml
* @param logDirectory The log directory.
* @param hasLogback Uses logback?
@@ -77,14 +79,13 @@ public class JavaCmdJobManagerDecorator extends AbstractKubernetesStepDecorator
*/
private static String getJobManagerStartCommand(
Configuration flinkConfig,
- int jobManagerMemoryMb,
+ ProcessMemorySpec jobManagerProcessSpec,
String configDirectory,
String logDirectory,
boolean hasLogback,
boolean hasLog4j,
String mainClass) {
- final int heapSize = BootstrapTools.calculateHeapSize(jobManagerMemoryMb, flinkConfig);
- final String jvmMemOpts = String.format("-Xms%sm -Xmx%sm", heapSize, heapSize);
+ final String jvmMemOpts = ProcessMemoryUtils.generateJvmParametersStr(jobManagerProcessSpec);
return KubernetesUtils.getCommonStartCommand(
flinkConfig,
KubernetesUtils.ClusterComponent.JOB_MANAGER,
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesTestBase.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesTestBase.java
index 5b2e4cd..7f88981 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesTestBase.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesTestBase.java
@@ -20,6 +20,8 @@ package org.apache.flink.kubernetes;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.configuration.MemorySize;
import org.apache.flink.core.testutils.CommonTestUtils;
import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
import org.apache.flink.kubernetes.kubeclient.Fabric8FlinkKubeClient;
@@ -51,6 +53,7 @@ public class KubernetesTestBase extends TestLogger {
protected static final String CONTAINER_IMAGE = "flink-k8s-test:latest";
protected static final KubernetesConfigOptions.ImagePullPolicy CONTAINER_IMAGE_PULL_POLICY =
KubernetesConfigOptions.ImagePullPolicy.IfNotPresent;
+ protected static final int JOB_MANAGER_MEMORY = 768;
@Rule
public MixedKubernetesServer server = new MixedKubernetesServer(true, true);
@@ -74,6 +77,7 @@ public class KubernetesTestBase extends TestLogger {
flinkConfig.setString(KubernetesConfigOptions.CLUSTER_ID, CLUSTER_ID);
flinkConfig.setString(KubernetesConfigOptions.CONTAINER_IMAGE, CONTAINER_IMAGE);
flinkConfig.set(KubernetesConfigOptions.CONTAINER_IMAGE_PULL_POLICY, CONTAINER_IMAGE_PULL_POLICY);
+ flinkConfig.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(JOB_MANAGER_MEMORY));
flinkConfDir = temporaryFolder.newFolder().getAbsoluteFile();
hadoopConfDir = temporaryFolder.newFolder().getAbsoluteFile();
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/cli/KubernetesSessionCliTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/cli/KubernetesSessionCliTest.java
index a7bf478..df7fe6a 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/cli/KubernetesSessionCliTest.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/cli/KubernetesSessionCliTest.java
@@ -84,7 +84,7 @@ public class KubernetesSessionCliTest {
"-e", KubernetesSessionClusterExecutor.NAME,
"-D" + TaskManagerOptions.NUM_TASK_SLOTS.key() + "=3"};
- final KubernetesSessionCli cli = createFlinkKubernetesCustomCliWithTmTotalMemory(1234);
+ final KubernetesSessionCli cli = createFlinkKubernetesCustomCliWithJmAndTmTotalMemory(1234);
final Configuration executorConfig = cli.getEffectiveConfiguration(params);
final ClusterClientFactory<String> clientFactory = getClusterClientFactory(executorConfig);
@@ -96,7 +96,7 @@ public class KubernetesSessionCliTest {
@Test
public void testResumeFromKubernetesID() throws Exception {
- final KubernetesSessionCli cli = createFlinkKubernetesCustomCliWithTmTotalMemory(1024);
+ final KubernetesSessionCli cli = createFlinkKubernetesCustomCliWithJmAndTmTotalMemory(1024);
final String clusterId = "my-test-CLUSTER_ID";
final String[] args = new String[] {
@@ -120,13 +120,13 @@ public class KubernetesSessionCliTest {
final int taskManagerMemory = 7331;
final int slotsPerTaskManager = 30;
- configuration.set(JobManagerOptions.JOB_MANAGER_HEAP_MEMORY, MemorySize.ofMebiBytes(jobManagerMemory));
+ configuration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(jobManagerMemory));
configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(taskManagerMemory));
configuration.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, slotsPerTaskManager);
final String[] args = {
"-e", KubernetesSessionClusterExecutor.NAME,
- "-D" + JobManagerOptions.JOB_MANAGER_HEAP_MEMORY.key() + "=" + jobManagerMemory + "m",
+ "-D" + JobManagerOptions.TOTAL_PROCESS_MEMORY.key() + "=" + jobManagerMemory + "m",
"-D" + TaskManagerOptions.TOTAL_PROCESS_MEMORY.key() + "=" + taskManagerMemory + "m",
"-D" + TaskManagerOptions.NUM_TASK_SLOTS.key() + "=" + slotsPerTaskManager
};
@@ -150,7 +150,7 @@ public class KubernetesSessionCliTest {
public void testConfigurationClusterSpecification() throws Exception {
final Configuration configuration = new Configuration();
final int jobManagerMemory = 1337;
- configuration.set(JobManagerOptions.JOB_MANAGER_HEAP_MEMORY, MemorySize.ofMebiBytes(jobManagerMemory));
+ configuration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(jobManagerMemory));
final int taskManagerMemory = 7331;
configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(taskManagerMemory));
final int slotsPerTaskManager = 42;
@@ -175,11 +175,11 @@ public class KubernetesSessionCliTest {
public void testHeapMemoryPropertyWithUnitMB() throws Exception {
final String[] args = new String[] {
"-e", KubernetesSessionClusterExecutor.NAME,
- "-D" + JobManagerOptions.JOB_MANAGER_HEAP_MEMORY.key() + "=1024m",
+ "-D" + JobManagerOptions.TOTAL_PROCESS_MEMORY.key() + "=1024m",
"-D" + TaskManagerOptions.TOTAL_PROCESS_MEMORY.key() + "=2048m"
};
- final KubernetesSessionCli cli = createFlinkKubernetesCustomCliWithTmTotalMemory(1024);
+ final KubernetesSessionCli cli = createFlinkKubernetesCustomCliWithJmAndTmTotalMemory(1024);
final Configuration executorConfig = cli.getEffectiveConfiguration(args);
final ClusterClientFactory<String> clientFactory = getClusterClientFactory(executorConfig);
@@ -196,11 +196,11 @@ public class KubernetesSessionCliTest {
public void testHeapMemoryPropertyWithArbitraryUnit() throws Exception {
final String[] args = new String[] {
"-e", KubernetesSessionClusterExecutor.NAME,
- "-D" + JobManagerOptions.JOB_MANAGER_HEAP_MEMORY.key() + "=1g",
+ "-D" + JobManagerOptions.TOTAL_PROCESS_MEMORY.key() + "=1g",
"-D" + TaskManagerOptions.TOTAL_PROCESS_MEMORY.key() + "=3g"
};
- final KubernetesSessionCli cli = createFlinkKubernetesCustomCliWithTmTotalMemory(1024);
+ final KubernetesSessionCli cli = createFlinkKubernetesCustomCliWithJmAndTmTotalMemory(1024);
final Configuration executorConfig = cli.getEffectiveConfiguration(args);
final ClusterClientFactory<String> clientFactory = getClusterClientFactory(executorConfig);
@@ -239,7 +239,7 @@ public class KubernetesSessionCliTest {
"-e", KubernetesSessionClusterExecutor.NAME
};
- final KubernetesSessionCli cli = createFlinkKubernetesCustomCliWithTmTotalMemory(1024);
+ final KubernetesSessionCli cli = createFlinkKubernetesCustomCliWithJmAndTmTotalMemory(1024);
final Configuration executorConfig = cli.getEffectiveConfiguration(args);
final ClusterClientFactory<String> clientFactory = getClusterClientFactory(executorConfig);
@@ -254,8 +254,9 @@ public class KubernetesSessionCliTest {
return clusterClientServiceLoader.getClusterClientFactory(executorConfig);
}
- private KubernetesSessionCli createFlinkKubernetesCustomCliWithTmTotalMemory(int totalMemory) {
+ private KubernetesSessionCli createFlinkKubernetesCustomCliWithJmAndTmTotalMemory(int totalMemory) {
Configuration configuration = new Configuration();
+ configuration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(totalMemory));
configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(totalMemory));
return new KubernetesSessionCli(configuration);
}
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/KubernetesJobManagerTestBase.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/KubernetesJobManagerTestBase.java
index 6d2dd7c..2e12bd3 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/KubernetesJobManagerTestBase.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/KubernetesJobManagerTestBase.java
@@ -21,6 +21,7 @@ package org.apache.flink.kubernetes.kubeclient;
import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.configuration.BlobServerOptions;
import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.ResourceManagerOptions;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.kubernetes.KubernetesTestBase;
@@ -83,6 +84,7 @@ public class KubernetesJobManagerTestBase extends KubernetesTestBase {
this.flinkConfig.setString(ResourceManagerOptions.CONTAINERIZED_MASTER_ENV_PREFIX + k, v));
this.flinkConfig.set(KubernetesConfigOptions.JOB_MANAGER_LABELS, userLabels);
this.flinkConfig.set(KubernetesConfigOptions.JOB_MANAGER_NODE_SELECTOR, nodeSelector);
+ this.flinkConfig.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(JOB_MANAGER_MEMORY));
final ClusterSpecification clusterSpecification = new ClusterSpecification.ClusterSpecificationBuilder()
.setMasterMemoryMB(JOB_MANAGER_MEMORY)
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/JavaCmdJobManagerDecoratorTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/JavaCmdJobManagerDecoratorTest.java
index 36ccd1e..ea7c309 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/JavaCmdJobManagerDecoratorTest.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/JavaCmdJobManagerDecoratorTest.java
@@ -25,6 +25,8 @@ import org.apache.flink.kubernetes.configuration.KubernetesConfigOptionsInternal
import org.apache.flink.kubernetes.entrypoint.KubernetesSessionClusterEntrypoint;
import org.apache.flink.kubernetes.kubeclient.FlinkPod;
import org.apache.flink.kubernetes.kubeclient.KubernetesJobManagerTestBase;
+import org.apache.flink.runtime.jobmanager.JobManagerProcessUtils;
+import org.apache.flink.runtime.util.config.memory.ProcessMemoryUtils;
import io.fabric8.kubernetes.api.model.Container;
import org.junit.Before;
@@ -63,8 +65,8 @@ public class JavaCmdJobManagerDecoratorTest extends KubernetesJobManagerTestBase
FLINK_LOG_DIR_IN_POD, FLINK_LOG_DIR_IN_POD);
// Memory variables
- private static final String jmJvmMem = String.format("-Xms%dm -Xmx%dm",
- JOB_MANAGER_MEMORY - 600, JOB_MANAGER_MEMORY - 600);
+ private static final String jmJvmMem = ProcessMemoryUtils.generateJvmParametersStr(
+ JobManagerProcessUtils.createDefaultJobManagerProcessSpec(JOB_MANAGER_MEMORY));
private JavaCmdJobManagerDecorator javaCmdJobManagerDecorator;
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesJobManagerParametersTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesJobManagerParametersTest.java
index 00f3bae..1f6dc00 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesJobManagerParametersTest.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesJobManagerParametersTest.java
@@ -47,7 +47,6 @@ import static org.junit.Assert.fail;
*/
public class KubernetesJobManagerParametersTest extends KubernetesTestBase {
- private static final int JOB_MANAGER_MEMORY = 768;
private static final double JOB_MANAGER_CPU = 2.0;
private final ClusterSpecification clusterSpecification = new ClusterSpecification.ClusterSpecificationBuilder()
[flink] 10/17: [FLINK-16745][yarn] Start Yarn JM with FLIP-116 JVM
memory args
Posted by az...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
azagrebin pushed a commit to branch flip116
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 1a831ec08b87f98b856874e6ab69c7870ea2e601
Author: Andrey Zagrebin <az...@apache.org>
AuthorDate: Wed Apr 8 18:06:58 2020 +0300
[FLINK-16745][yarn] Start Yarn JM with FLIP-116 JVM memory args
---
.../flink/yarn/YARNFileReplicationITCase.java | 5 +++-
.../flink/yarn/YARNHighAvailabilityITCase.java | 4 ++-
.../java/org/apache/flink/yarn/YARNITCase.java | 5 +++-
.../apache/flink/yarn/YarnConfigurationITCase.java | 5 +++-
.../apache/flink/yarn/YarnClusterDescriptor.java | 10 ++++---
.../apache/flink/yarn/cli/FlinkYarnSessionCli.java | 2 +-
.../apache/flink/yarn/FlinkYarnSessionCliTest.java | 33 ++++++++++----------
.../flink/yarn/YarnClusterDescriptorTest.java | 35 ++++++++++++----------
8 files changed, 58 insertions(+), 41 deletions(-)
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNFileReplicationITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNFileReplicationITCase.java
index 055380b..993adbf 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNFileReplicationITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNFileReplicationITCase.java
@@ -22,6 +22,7 @@ import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.configuration.AkkaOptions;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.jobgraph.JobGraph;
@@ -84,8 +85,9 @@ public class YARNFileReplicationITCase extends YarnTestBase {
yarnClusterDescriptor.addShipFiles(Arrays.asList(flinkLibFolder.listFiles()));
yarnClusterDescriptor.addShipFiles(Arrays.asList(flinkShadedHadoopDir.listFiles()));
+ final int masterMemory = yarnClusterDescriptor.getFlinkConfiguration().get(JobManagerOptions.TOTAL_PROCESS_MEMORY).getMebiBytes();
final ClusterSpecification clusterSpecification = new ClusterSpecification.ClusterSpecificationBuilder()
- .setMasterMemoryMB(768)
+ .setMasterMemoryMB(masterMemory)
.setTaskManagerMemoryMB(1024)
.setSlotsPerTaskManager(1)
.createClusterSpecification();
@@ -130,6 +132,7 @@ public class YARNFileReplicationITCase extends YarnTestBase {
private Configuration getDefaultConfiguration() {
final Configuration configuration = new Configuration();
+ configuration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(768));
configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.parse("1g"));
configuration.setString(AkkaOptions.ASK_TIMEOUT, "30 s");
configuration.setString(CLASSPATH_INCLUDE_USER_JAR, YarnConfigOptions.UserJarInclusion.DISABLED.toString());
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
index 6420078..b4ec5ea 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
@@ -26,6 +26,7 @@ import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.HighAvailabilityOptions;
+import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.ResourceManagerOptions;
import org.apache.flink.configuration.RestartStrategyOptions;
@@ -273,6 +274,7 @@ public class YARNHighAvailabilityITCase extends YarnTestBase {
@Nonnull
private YarnClusterDescriptor setupYarnClusterDescriptor() {
final Configuration flinkConfiguration = new Configuration();
+ flinkConfiguration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(768));
flinkConfiguration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.parse("1g"));
flinkConfiguration.setString(YarnConfigOptions.APPLICATION_ATTEMPTS, "10");
flinkConfiguration.setString(HighAvailabilityOptions.HA_MODE, "zookeeper");
@@ -290,7 +292,7 @@ public class YARNHighAvailabilityITCase extends YarnTestBase {
}
private RestClusterClient<ApplicationId> deploySessionCluster(YarnClusterDescriptor yarnClusterDescriptor) throws ClusterDeploymentException {
- final int masterMemory = 256;
+ final int masterMemory = yarnClusterDescriptor.getFlinkConfiguration().get(JobManagerOptions.TOTAL_PROCESS_MEMORY).getMebiBytes();
final int taskManagerMemory = 1024;
final ClusterClient<ApplicationId> yarnClusterClient = yarnClusterDescriptor
.deploySessionCluster(new ClusterSpecification.ClusterSpecificationBuilder()
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNITCase.java
index bb15e28..b7139af 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNITCase.java
@@ -22,6 +22,7 @@ import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.configuration.AkkaOptions;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.jobgraph.JobGraph;
@@ -94,8 +95,9 @@ public class YARNITCase extends YarnTestBase {
yarnClusterDescriptor.addShipFiles(Arrays.asList(flinkLibFolder.listFiles()));
yarnClusterDescriptor.addShipFiles(Arrays.asList(flinkShadedHadoopDir.listFiles()));
+ final int masterMemory = yarnClusterDescriptor.getFlinkConfiguration().get(JobManagerOptions.TOTAL_PROCESS_MEMORY).getMebiBytes();
final ClusterSpecification clusterSpecification = new ClusterSpecification.ClusterSpecificationBuilder()
- .setMasterMemoryMB(768)
+ .setMasterMemoryMB(masterMemory)
.setTaskManagerMemoryMB(1024)
.setSlotsPerTaskManager(1)
.createClusterSpecification();
@@ -138,6 +140,7 @@ public class YARNITCase extends YarnTestBase {
private Configuration createDefaultConfiguration(YarnConfigOptions.UserJarInclusion userJarInclusion) {
Configuration configuration = new Configuration();
+ configuration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(768));
configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.parse("1g"));
configuration.setString(AkkaOptions.ASK_TIMEOUT, "30 s");
configuration.setString(CLASSPATH_INCLUDE_USER_JAR, userJarInclusion.toString());
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java
index d44d612..ee538d4 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java
@@ -26,6 +26,8 @@ import org.apache.flink.client.program.PackagedProgram;
import org.apache.flink.client.program.PackagedProgramUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ResourceManagerOptions;
+import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.clusterframework.TaskExecutorProcessSpec;
import org.apache.flink.runtime.clusterframework.TaskExecutorProcessUtils;
@@ -88,9 +90,10 @@ public class YarnConfigurationITCase extends YarnTestBase {
final int slotsPerTaskManager = 3;
configuration.set(TaskManagerOptions.NUM_TASK_SLOTS, slotsPerTaskManager);
+ final int masterMemory = 768;
+ configuration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(masterMemory));
final TaskExecutorProcessSpec tmResourceSpec = TaskExecutorProcessUtils.processSpecFromConfig(configuration);
- final int masterMemory = 64;
final int taskManagerMemory = tmResourceSpec.getTotalProcessMemorySize().getMebiBytes();
final YarnConfiguration yarnConfiguration = getYarnConfiguration();
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java
index cde732b..daae7ed 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java
@@ -44,7 +44,10 @@ import org.apache.flink.runtime.clusterframework.BootstrapTools;
import org.apache.flink.runtime.entrypoint.ClusterEntrypoint;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobmanager.HighAvailabilityMode;
+import org.apache.flink.runtime.jobmanager.JobManagerProcessSpec;
+import org.apache.flink.runtime.jobmanager.JobManagerProcessUtils;
import org.apache.flink.runtime.util.HadoopUtils;
+import org.apache.flink.runtime.util.config.memory.ProcessMemoryUtils;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.ShutdownHookUtil;
@@ -932,7 +935,7 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
hasLogback,
hasLog4j,
hasKrb5,
- clusterSpecification.getMasterMemoryMB());
+ JobManagerProcessUtils.processSpecFromConfig(flinkConfiguration));
// setup security tokens
if (UserGroupInformation.isSecurityEnabled()) {
@@ -1570,7 +1573,7 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
boolean hasLogback,
boolean hasLog4j,
boolean hasKrb5,
- int jobManagerMemoryMb) {
+ JobManagerProcessSpec processSpec) {
// ------------------ Prepare Application Master Container ------------------------------
// respect custom JVM options in the YAML file
@@ -1590,8 +1593,7 @@ public class YarnClusterDescriptor implements ClusterDescriptor<ApplicationId> {
final Map<String, String> startCommandValues = new HashMap<>();
startCommandValues.put("java", "$JAVA_HOME/bin/java");
- int heapSize = BootstrapTools.calculateHeapSize(jobManagerMemoryMb, flinkConfiguration);
- String jvmHeapMem = String.format("-Xms%sm -Xmx%sm", heapSize, heapSize);
+ String jvmHeapMem = ProcessMemoryUtils.generateJvmParametersStr(processSpec);
startCommandValues.put("jvmmem", jvmHeapMem);
startCommandValues.put("jvmopts", javaOpts);
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java b/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
index 352ca6e..4ca49c6 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/cli/FlinkYarnSessionCli.java
@@ -355,7 +355,7 @@ public class FlinkYarnSessionCli extends AbstractCustomCommandLine {
if (!MemorySize.MemoryUnit.hasUnit(jmMemoryVal)) {
jmMemoryVal += "m";
}
- effectiveConfiguration.set(JobManagerOptions.JOB_MANAGER_HEAP_MEMORY, MemorySize.parse(jmMemoryVal));
+ effectiveConfiguration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.parse(jmMemoryVal));
}
if (commandLine.hasOption(tmMemory.getOpt())) {
diff --git a/flink-yarn/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java b/flink-yarn/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java
index bf097fb..6657190 100644
--- a/flink-yarn/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java
+++ b/flink-yarn/src/test/java/org/apache/flink/yarn/FlinkYarnSessionCliTest.java
@@ -111,7 +111,7 @@ public class FlinkYarnSessionCliTest extends TestLogger {
String[] params =
new String[] {"-ys", "3"};
- FlinkYarnSessionCli yarnCLI = createFlinkYarnSessionCliWithTmTotalMemory(2048);
+ FlinkYarnSessionCli yarnCLI = createFlinkYarnSessionCliWithJmAndTmTotalMemory(2048);
final CommandLine commandLine = yarnCLI.parseCommandLineOptions(params, true);
@@ -314,7 +314,7 @@ public class FlinkYarnSessionCliTest extends TestLogger {
final int taskManagerMemory = 7331;
final int slotsPerTaskManager = 30;
- configuration.set(JobManagerOptions.JOB_MANAGER_HEAP_MEMORY, MemorySize.ofMebiBytes(jobManagerMemory));
+ configuration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(jobManagerMemory));
configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(taskManagerMemory));
configuration.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, slotsPerTaskManager);
@@ -340,7 +340,7 @@ public class FlinkYarnSessionCliTest extends TestLogger {
public void testConfigurationClusterSpecification() throws Exception {
final Configuration configuration = new Configuration();
final int jobManagerMemory = 1337;
- configuration.set(JobManagerOptions.JOB_MANAGER_HEAP_MEMORY, MemorySize.ofMebiBytes(jobManagerMemory));
+ configuration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(jobManagerMemory));
final int taskManagerMemory = 7331;
configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(taskManagerMemory));
final int slotsPerTaskManager = 42;
@@ -361,10 +361,10 @@ public class FlinkYarnSessionCliTest extends TestLogger {
}
/**
- * Tests the specifying heap memory without unit for job manager and task manager.
+ * Tests the specifying total process memory without unit for job manager and task manager.
*/
@Test
- public void testHeapMemoryPropertyWithoutUnit() throws Exception {
+ public void testMemoryPropertyWithoutUnit() throws Exception {
final String[] args = new String[] { "-yjm", "1024", "-ytm", "2048" };
final FlinkYarnSessionCli flinkYarnSessionCli = createFlinkYarnSessionCli();
@@ -379,10 +379,10 @@ public class FlinkYarnSessionCliTest extends TestLogger {
}
/**
- * Tests the specifying heap memory with unit (MB) for job manager and task manager.
+ * Tests the specifying total process memory with unit (MB) for job manager and task manager.
*/
@Test
- public void testHeapMemoryPropertyWithUnitMB() throws Exception {
+ public void testMemoryPropertyWithUnitMB() throws Exception {
final String[] args = new String[] { "-yjm", "1024m", "-ytm", "2048m" };
final FlinkYarnSessionCli flinkYarnSessionCli = createFlinkYarnSessionCli();
final CommandLine commandLine = flinkYarnSessionCli.parseCommandLineOptions(args, false);
@@ -396,10 +396,10 @@ public class FlinkYarnSessionCliTest extends TestLogger {
}
/**
- * Tests the specifying heap memory with arbitrary unit for job manager and task manager.
+ * Tests the specifying total process memory with arbitrary unit for job manager and task manager.
*/
@Test
- public void testHeapMemoryPropertyWithArbitraryUnit() throws Exception {
+ public void testMemoryPropertyWithArbitraryUnit() throws Exception {
final String[] args = new String[] { "-yjm", "1g", "-ytm", "2g" };
final FlinkYarnSessionCli flinkYarnSessionCli = createFlinkYarnSessionCli();
final CommandLine commandLine = flinkYarnSessionCli.parseCommandLineOptions(args, false);
@@ -434,12 +434,12 @@ public class FlinkYarnSessionCliTest extends TestLogger {
}
/**
- * Tests the specifying job manager heap memory with config default value for job manager and task manager.
+ * Tests the specifying job manager total process memory with config default value for job manager and task manager.
*/
@Test
- public void testJobManagerHeapMemoryPropertyWithConfigDefaultValue() throws Exception {
- int tmMemory = 2048;
- final FlinkYarnSessionCli flinkYarnSessionCli = createFlinkYarnSessionCliWithTmTotalMemory(tmMemory);
+ public void testJobManagerMemoryPropertyWithConfigDefaultValue() throws Exception {
+ int procMemory = 2048;
+ final FlinkYarnSessionCli flinkYarnSessionCli = createFlinkYarnSessionCliWithJmAndTmTotalMemory(procMemory);
final CommandLine commandLine = flinkYarnSessionCli.parseCommandLineOptions(new String[0], false);
@@ -447,8 +447,8 @@ public class FlinkYarnSessionCliTest extends TestLogger {
final ClusterClientFactory<ApplicationId> clientFactory = getClusterClientFactory(executorConfig);
final ClusterSpecification clusterSpecification = clientFactory.getClusterSpecification(executorConfig);
- assertThat(clusterSpecification.getMasterMemoryMB(), is(1024));
- assertThat(clusterSpecification.getTaskManagerMemoryMB(), is(tmMemory));
+ assertThat(clusterSpecification.getMasterMemoryMB(), is(procMemory));
+ assertThat(clusterSpecification.getTaskManagerMemoryMB(), is(procMemory));
}
@Test
@@ -490,8 +490,9 @@ public class FlinkYarnSessionCliTest extends TestLogger {
return createFlinkYarnSessionCli(new Configuration());
}
- private FlinkYarnSessionCli createFlinkYarnSessionCliWithTmTotalMemory(int totalMemomory) throws FlinkException {
+ private FlinkYarnSessionCli createFlinkYarnSessionCliWithJmAndTmTotalMemory(int totalMemomory) throws FlinkException {
Configuration configuration = new Configuration();
+ configuration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(totalMemomory));
configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(totalMemomory));
return createFlinkYarnSessionCli(configuration);
}
diff --git a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java
index 7a855e9..09d2133 100644
--- a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java
+++ b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java
@@ -26,6 +26,8 @@ import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.configuration.ResourceManagerOptions;
import org.apache.flink.core.testutils.CommonTestUtils;
+import org.apache.flink.runtime.jobmanager.JobManagerProcessSpec;
+import org.apache.flink.runtime.util.config.memory.ProcessMemoryUtils;
import org.apache.flink.util.TestLogger;
import org.apache.flink.yarn.cli.FlinkYarnSessionCli;
import org.apache.flink.yarn.configuration.YarnConfigOptions;
@@ -55,6 +57,7 @@ import java.util.Map;
import java.util.Set;
import static junit.framework.TestCase.assertTrue;
+import static org.apache.flink.runtime.jobmanager.JobManagerProcessUtils.createDefaultJobManagerProcessSpec;
import static org.apache.flink.yarn.configuration.YarnConfigOptions.CLASSPATH_INCLUDE_USER_JAR;
import static org.hamcrest.Matchers.containsString;
import static org.junit.Assert.assertEquals;
@@ -156,8 +159,9 @@ public class YarnClusterDescriptorTest extends TestLogger {
Configuration cfg = new Configuration();
YarnClusterDescriptor clusterDescriptor = createYarnClusterDescriptor(cfg);
+ final JobManagerProcessSpec jobManagerProcessSpec = createDefaultJobManagerProcessSpec(1024);
final String java = "$JAVA_HOME/bin/java";
- final String jvmmem = "-Xms424m -Xmx424m";
+ final String jvmmem = ProcessMemoryUtils.generateJvmParametersStr(jobManagerProcessSpec);
final String jvmOpts = "-Djvm"; // if set
final String jmJvmOpts = "-DjmJvm"; // if set
final String krb5 = "-Djava.security.krb5.conf=krb5.conf";
@@ -173,7 +177,6 @@ public class YarnClusterDescriptorTest extends TestLogger {
final String redirects =
"1> " + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/jobmanager.out " +
"2> " + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/jobmanager.err";
- final int jobManagerMemory = 1024;
try {
// no logging, with/out krb5
@@ -188,7 +191,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
false,
false,
false,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
assertEquals(
@@ -202,7 +205,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
false,
false,
true,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
// logback only, with/out krb5
@@ -217,7 +220,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
true,
false,
false,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
assertEquals(
@@ -231,7 +234,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
true,
false,
true,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
// log4j, with/out krb5
@@ -246,7 +249,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
false,
true,
false,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
assertEquals(
@@ -260,7 +263,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
false,
true,
true,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
// logback + log4j, with/out krb5
@@ -275,7 +278,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
true,
true,
false,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
assertEquals(
@@ -289,7 +292,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
true,
true,
true,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
// logback + log4j, with/out krb5, different JVM opts
@@ -307,7 +310,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
true,
true,
false,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
assertEquals(
@@ -321,7 +324,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
true,
true,
true,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
// logback + log4j, with/out krb5, different JVM opts
@@ -338,7 +341,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
true,
true,
false,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
assertEquals(
@@ -352,7 +355,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
true,
true,
true,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
// now try some configurations with different yarn.container-start-command-template
@@ -370,7 +373,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
true,
true,
true,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
cfg.setString(ConfigConstants.YARN_CONTAINER_START_COMMAND_TEMPLATE,
@@ -388,7 +391,7 @@ public class YarnClusterDescriptorTest extends TestLogger {
true,
true,
true,
- jobManagerMemory)
+ jobManagerProcessSpec)
.getCommands().get(0));
} finally {
clusterDescriptor.close();
[flink] 07/17: [FLINK-16472][dist] Remove unused memory
configuration logics from config.sh.
Posted by az...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
azagrebin pushed a commit to branch flip116
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d3bc2a6b3f351776f883a40d10ec5f4c44a82ab6
Author: Xintong Song <to...@gmail.com>
AuthorDate: Wed Apr 1 19:44:28 2020 +0800
[FLINK-16472][dist] Remove unused memory configuration logics from config.sh.
---
flink-dist/src/main/flink-bin/bin/config.sh | 152 +---------------------------
1 file changed, 1 insertion(+), 151 deletions(-)
diff --git a/flink-dist/src/main/flink-bin/bin/config.sh b/flink-dist/src/main/flink-bin/bin/config.sh
index 08c709d..a40a1d4 100755
--- a/flink-dist/src/main/flink-bin/bin/config.sh
+++ b/flink-dist/src/main/flink-bin/bin/config.sh
@@ -119,9 +119,6 @@ DEFAULT_HADOOP_CONF_DIR="" # Hadoop Configuration Direc
# CONFIG KEYS: The default values can be overwritten by the following keys in conf/flink-conf.yaml
########################################################################################################################
-KEY_JOBM_MEM_SIZE="jobmanager.heap.size"
-KEY_JOBM_MEM_MB="jobmanager.heap.mb"
-
KEY_TASKM_COMPUTE_NUMA="taskmanager.compute.numa"
KEY_ENV_PID_DIR="env.pid.dir"
@@ -140,143 +137,6 @@ KEY_HIGH_AVAILABILITY="high-availability"
KEY_ZK_HEAP_MB="zookeeper.heap.mb"
########################################################################################################################
-# MEMORY SIZE UNIT
-########################################################################################################################
-
-BYTES_UNITS=("b" "bytes")
-KILO_BYTES_UNITS=("k" "kb" "kibibytes")
-MEGA_BYTES_UNITS=("m" "mb" "mebibytes")
-GIGA_BYTES_UNITS=("g" "gb" "gibibytes")
-TERA_BYTES_UNITS=("t" "tb" "tebibytes")
-
-hasUnit() {
- text=$1
-
- trimmed=$(echo -e "${text}" | tr -d '[:space:]')
-
- if [ -z "$trimmed" -o "$trimmed" == " " ]; then
- echo "$trimmed is an empty- or whitespace-only string"
- exit 1
- fi
-
- len=${#trimmed}
- pos=0
-
- while [ $pos -lt $len ]; do
- current=${trimmed:pos:1}
- if [[ ! $current < '0' ]] && [[ ! $current > '9' ]]; then
- let pos+=1
- else
- break
- fi
- done
-
- number=${trimmed:0:pos}
-
- unit=${trimmed:$pos}
- unit=$(echo -e "${unit}" | tr -d '[:space:]')
- unit=$(echo -e "${unit}" | tr '[A-Z]' '[a-z]')
-
- [[ ! -z "$unit" ]]
-}
-
-parseBytes() {
- text=$1
-
- trimmed=$(echo -e "${text}" | tr -d '[:space:]')
-
- if [ -z "$trimmed" -o "$trimmed" == " " ]; then
- echo "$trimmed is an empty- or whitespace-only string"
- exit 1
- fi
-
- len=${#trimmed}
- pos=0
-
- while [ $pos -lt $len ]; do
- current=${trimmed:pos:1}
- if [[ ! $current < '0' ]] && [[ ! $current > '9' ]]; then
- let pos+=1
- else
- break
- fi
- done
-
- number=${trimmed:0:pos}
-
- unit=${trimmed:$pos}
- unit=$(echo -e "${unit}" | tr -d '[:space:]')
- unit=$(echo -e "${unit}" | tr '[A-Z]' '[a-z]')
-
- if [ -z "$number" ]; then
- echo "text does not start with a number"
- exit 1
- fi
-
- local multiplier
- if [ -z "$unit" ]; then
- multiplier=1
- else
- if matchesAny $unit "${BYTES_UNITS[*]}"; then
- multiplier=1
- elif matchesAny $unit "${KILO_BYTES_UNITS[*]}"; then
- multiplier=1024
- elif matchesAny $unit "${MEGA_BYTES_UNITS[*]}"; then
- multiplier=`expr 1024 \* 1024`
- elif matchesAny $unit "${GIGA_BYTES_UNITS[*]}"; then
- multiplier=`expr 1024 \* 1024 \* 1024`
- elif matchesAny $unit "${TERA_BYTES_UNITS[*]}"; then
- multiplier=`expr 1024 \* 1024 \* 1024 \* 1024`
- else
- echo "[ERROR] Memory size unit $unit does not match any of the recognized units"
- exit 1
- fi
- fi
-
- ((result=$number * $multiplier))
-
- if [ $[result / multiplier] != "$number" ]; then
- echo "[ERROR] The value $text cannot be re represented as 64bit number of bytes (numeric overflow)."
- exit 1
- fi
-
- echo "$result"
-}
-
-matchesAny() {
- str=$1
- variants=$2
-
- for s in ${variants[*]}; do
- if [ $str == $s ]; then
- return 0
- fi
- done
-
- return 1
-}
-
-getKibiBytes() {
- bytes=$1
- echo "$(($bytes >>10))"
-}
-
-getMebiBytes() {
- bytes=$1
- echo "$(($bytes >> 20))"
-}
-
-getGibiBytes() {
- bytes=$1
- echo "$(($bytes >> 30))"
-}
-
-getTebiBytes() {
- bytes=$1
- echo "$(($bytes >> 40))"
-}
-
-########################################################################################################################
# PATHS AND CONFIG
########################################################################################################################
@@ -366,16 +226,6 @@ fi
IS_NUMBER="^[0-9]+$"
-# Define FLINK_JM_HEAP if it is not already set
-if [ -z "${FLINK_JM_HEAP}" ]; then
- FLINK_JM_HEAP=$(readFromConfig ${KEY_JOBM_MEM_SIZE} 0 "${YAML_CONF}")
-fi
-
-# Try read old config key, if new key not exists
-if [ "${FLINK_JM_HEAP}" == 0 ]; then
- FLINK_JM_HEAP_MB=$(readFromConfig ${KEY_JOBM_MEM_MB} 0 "${YAML_CONF}")
-fi
-
# Verify that NUMA tooling is available
command -v numactl >/dev/null 2>&1
if [[ $? -ne 0 ]]; then
@@ -466,7 +316,7 @@ fi
# Arguments for the JVM. Used for job and task manager JVMs.
# DO NOT USE FOR MEMORY SETTINGS! Use conf/flink-conf.yaml with keys
-# KEY_JOBM_MEM_SIZE and TaskManagerOptions#TOTAL_PROCESS_MEMORY for that!
+# JobManagerOptions#TOTAL_PROCESS_MEMORY and TaskManagerOptions#TOTAL_PROCESS_MEMORY for that!
if [ -z "${JVM_ARGS}" ]; then
JVM_ARGS=""
fi
[flink] 02/17: [FLINK-16874] Respect the dynamic options when
calculating memory options in taskmanager.sh
Posted by az...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
azagrebin pushed a commit to branch flip116
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 8d60c0e06c0256b858a5876fe7c8add0a506e6f7
Author: Yangze Guo <ka...@gmail.com>
AuthorDate: Tue Mar 31 16:22:25 2020 +0800
[FLINK-16874] Respect the dynamic options when calculating memory options in taskmanager.sh
---
flink-dist/src/main/flink-bin/bin/config.sh | 5 +++--
flink-dist/src/main/flink-bin/bin/taskmanager.sh | 2 +-
flink-dist/src/test/bin/runBashJavaUtilsCmd.sh | 6 ++++--
.../java/org/apache/flink/dist/BashJavaUtilsITCase.java | 13 +++++++++++++
4 files changed, 21 insertions(+), 5 deletions(-)
diff --git a/flink-dist/src/main/flink-bin/bin/config.sh b/flink-dist/src/main/flink-bin/bin/config.sh
index 1258318..6b9ba7c 100755
--- a/flink-dist/src/main/flink-bin/bin/config.sh
+++ b/flink-dist/src/main/flink-bin/bin/config.sh
@@ -626,10 +626,11 @@ TMSlaves() {
runBashJavaUtilsCmd() {
local cmd=$1
local conf_dir=$2
- local class_path="${3:-$FLINK_BIN_DIR/bash-java-utils.jar:`findFlinkDistJar`}"
+ local class_path=$3
+ local dynamic_args=${@:4}
class_path=`manglePathList ${class_path}`
- local output=`${JAVA_RUN} -classpath ${class_path} org.apache.flink.runtime.util.BashJavaUtils ${cmd} --configDir ${conf_dir} 2>&1 | tail -n 1000`
+ local output=`${JAVA_RUN} -classpath ${class_path} org.apache.flink.runtime.util.BashJavaUtils ${cmd} --configDir ${conf_dir} $dynamic_args 2>&1 | tail -n 1000`
if [[ $? -ne 0 ]]; then
echo "[ERROR] Cannot run BashJavaUtils to execute command ${cmd}." 1>&2
# Print the output in case the user redirect the log to console.
diff --git a/flink-dist/src/main/flink-bin/bin/taskmanager.sh b/flink-dist/src/main/flink-bin/bin/taskmanager.sh
index fd1e1d4..4b78a05 100755
--- a/flink-dist/src/main/flink-bin/bin/taskmanager.sh
+++ b/flink-dist/src/main/flink-bin/bin/taskmanager.sh
@@ -48,7 +48,7 @@ if [[ $STARTSTOP == "start" ]] || [[ $STARTSTOP == "start-foreground" ]]; then
# Startup parameters
- java_utils_output=$(runBashJavaUtilsCmd GET_TM_RESOURCE_PARAMS ${FLINK_CONF_DIR})
+ java_utils_output=$(runBashJavaUtilsCmd GET_TM_RESOURCE_PARAMS ${FLINK_CONF_DIR} $FLINK_BIN_DIR/bash-java-utils.jar:$(findFlinkDistJar) "${ARGS[@]}")
num_lines=$(echo "${java_utils_output}" | wc -l)
logging_output=$(echo "${java_utils_output}" | head -n $((${num_lines} - 2)))
diff --git a/flink-dist/src/test/bin/runBashJavaUtilsCmd.sh b/flink-dist/src/test/bin/runBashJavaUtilsCmd.sh
index 2a7f883..e7b79ec 100755
--- a/flink-dist/src/test/bin/runBashJavaUtilsCmd.sh
+++ b/flink-dist/src/test/bin/runBashJavaUtilsCmd.sh
@@ -18,9 +18,11 @@
################################################################################
# Wrapper script to compare the TM heap size calculation of config.sh with Java
-USAGE="Usage: runBashJavaUtilsCmd.sh <command>"
+USAGE="Usage: runBashJavaUtilsCmd.sh <command> [dynamic args...]"
COMMAND=$1
+shift
+DYNAMIC_OPTS=$@
if [[ -z "${COMMAND}" ]]; then
echo "$USAGE"
@@ -36,6 +38,6 @@ FLINK_DIST_JAR=`find $FLINK_TARGET_DIR -name 'flink-dist*.jar'`
. ${bin}/../../main/flink-bin/bin/config.sh > /dev/null
-output=$(runBashJavaUtilsCmd GET_TM_RESOURCE_PARAMS ${FLINK_CONF_DIR} "$FLINK_TARGET_DIR/bash-java-utils.jar:$FLINK_DIST_JAR}" | tail -n 2)
+output=$(runBashJavaUtilsCmd GET_TM_RESOURCE_PARAMS ${FLINK_CONF_DIR} "$FLINK_TARGET_DIR/bash-java-utils.jar:$FLINK_DIST_JAR}" $DYNAMIC_OPTS | tail -n 2)
extractExecutionParams "$(echo "$output" | head -n 1)"
extractExecutionParams "$(echo "$output" | tail -n 1)"
diff --git a/flink-dist/src/test/java/org/apache/flink/dist/BashJavaUtilsITCase.java b/flink-dist/src/test/java/org/apache/flink/dist/BashJavaUtilsITCase.java
index e0587f7..f76b9c4 100644
--- a/flink-dist/src/test/java/org/apache/flink/dist/BashJavaUtilsITCase.java
+++ b/flink-dist/src/test/java/org/apache/flink/dist/BashJavaUtilsITCase.java
@@ -19,12 +19,14 @@
package org.apache.flink.dist;
import org.apache.flink.configuration.ConfigurationUtils;
+import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.util.BashJavaUtils;
import org.junit.Test;
import java.util.Arrays;
import java.util.List;
+import java.util.Map;
import static org.hamcrest.core.Is.is;
import static org.junit.Assert.assertThat;
@@ -47,4 +49,15 @@ public class BashJavaUtilsITCase extends JavaBashTestBase {
ConfigurationUtils.parseJvmArgString(lines.get(0));
ConfigurationUtils.parseTmResourceDynamicConfigs(lines.get(1));
}
+
+ @Test
+ public void testConfigOverwrittenByDynamicOpts() throws Exception {
+ double cpuCores = 39.0;
+ String[] commands = {RUN_BASH_JAVA_UTILS_CMD_SCRIPT, BashJavaUtils.Command.GET_TM_RESOURCE_PARAMS.toString(), "-D" + TaskManagerOptions.CPU_CORES.key() + "=" + cpuCores};
+ List<String> lines = Arrays.asList(executeScript(commands).split(System.lineSeparator()));
+
+ assertThat(lines.size(), is(2));
+ Map<String, String> configs = ConfigurationUtils.parseTmResourceDynamicConfigs(lines.get(1));
+ assertThat(Double.valueOf(configs.get(TaskManagerOptions.CPU_CORES.key())), is(cpuCores));
+ }
}
[flink] 16/17: change type to cron
Posted by az...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
azagrebin pushed a commit to branch flip116
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 5adb32b2dc5f145866034498b6a3fc252e5ab1aa
Author: Andrey Zagrebin <az...@apache.org>
AuthorDate: Wed Apr 15 13:01:39 2020 +0300
change type to cron
---
.travis.yml | 26 ++++++++++++++------------
1 file changed, 14 insertions(+), 12 deletions(-)
diff --git a/.travis.yml b/.travis.yml
index 3b4d212..029e31d 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -74,6 +74,8 @@ notifications:
on_success: change
on_pull_requests: false
+type: cron
+
stages:
- name: compile
- name: test
@@ -85,53 +87,53 @@ jdk: "openjdk8"
jobs:
include:
# main profile
- - if: type in (pull_request, push)
+ - if: type in (pull_request, push, cron)
stage: compile
script: ./tools/travis_controller.sh compile
env: PROFILE="-Dhadoop.version=2.8.3 -Dinclude_hadoop_aws -Dscala-2.11"
name: compile
- - if: type in (pull_request, push)
+ - if: type in (pull_request, push, cron)
stage: test
script: ./tools/travis_controller.sh core
env: PROFILE="-Dhadoop.version=2.8.3 -Dinclude_hadoop_aws -Dscala-2.11"
name: core
- - if: type in (pull_request, push)
+ - if: type in (pull_request, push, cron)
script: ./tools/travis_controller.sh python
env: PROFILE="-Dhadoop.version=2.8.3 -Dinclude_hadoop_aws -Dscala-2.11"
name: python
- - if: type in (pull_request, push)
+ - if: type in (pull_request, push, cron)
script: ./tools/travis_controller.sh libraries
env: PROFILE="-Dhadoop.version=2.8.3 -Dinclude_hadoop_aws -Dscala-2.11"
name: libraries
- - if: type in (pull_request, push)
+ - if: type in (pull_request, push, cron)
script: ./tools/travis_controller.sh blink_planner
env: PROFILE="-Dhadoop.version=2.8.3 -Dinclude_hadoop_aws -Dscala-2.11"
name: blink_planner
- - if: type in (pull_request, push)
+ - if: type in (pull_request, push, cron)
script: ./tools/travis_controller.sh connectors
env: PROFILE="-Dhadoop.version=2.8.3 -Dinclude_hadoop_aws -Dscala-2.11"
name: connectors
- - if: type in (pull_request, push)
+ - if: type in (pull_request, push, cron)
script: ./tools/travis_controller.sh kafka/gelly
env: PROFILE="-Dhadoop.version=2.8.3 -Dinclude_hadoop_aws -Dscala-2.11"
name: kafka/gelly
- - if: type in (pull_request, push)
+ - if: type in (pull_request, push, cron)
script: ./tools/travis_controller.sh tests
env: PROFILE="-Dhadoop.version=2.8.3 -Dinclude_hadoop_aws -Dscala-2.11"
name: tests
- - if: type in (pull_request, push)
+ - if: type in (pull_request, push, cron)
script: ./tools/travis_controller.sh legacy_scheduler_core
env: PROFILE="-Dhadoop.version=2.8.3 -Dinclude_hadoop_aws -Dscala-2.11"
name: core - legacy_scheduler
- - if: type in (pull_request, push)
+ - if: type in (pull_request, push, cron)
script: ./tools/travis_controller.sh legacy_scheduler_tests
env: PROFILE="-Dhadoop.version=2.8.3 -Dinclude_hadoop_aws -Dscala-2.11"
name: tests - legacy_scheduler
- - if: type in (pull_request, push)
+ - if: type in (pull_request, push, cron)
script: ./tools/travis_controller.sh misc
env: PROFILE="-Dhadoop.version=2.8.3 -Dinclude_hadoop_aws -Dscala-2.11"
name: misc
- - if: type in (pull_request, push)
+ - if: type in (pull_request, push, cron)
stage: cleanup
script: ./tools/travis_controller.sh cleanup
env: PROFILE="-Dhadoop.version=2.8.3 -Dinclude_hadoop_aws -Dscala-2.11"
[flink] 04/17: [FLINK-16742][dist] config.sh does not assume
BashJavaUtils always return two lines of results.
Posted by az...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
azagrebin pushed a commit to branch flip116
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 317b6f50a163b22e6fce7fd35689ad212a438d75
Author: Xintong Song <to...@gmail.com>
AuthorDate: Thu Mar 26 15:37:22 2020 +0800
[FLINK-16742][dist] config.sh does not assume BashJavaUtils always return two lines of results.
---
flink-dist/src/main/flink-bin/bin/config.sh | 31 +++++++++++-------
flink-dist/src/main/flink-bin/bin/taskmanager.sh | 18 ++++-------
flink-dist/src/test/bin/runBashJavaUtilsCmd.sh | 9 +++---
...JavaUtilsCmd.sh => runExtractLoggingOutputs.sh} | 13 +++-----
.../org/apache/flink/dist/BashJavaUtilsITCase.java | 37 +++++++++++++++++++---
.../apache/flink/runtime/util/BashJavaUtils.java | 4 ++-
6 files changed, 69 insertions(+), 43 deletions(-)
diff --git a/flink-dist/src/main/flink-bin/bin/config.sh b/flink-dist/src/main/flink-bin/bin/config.sh
index 6b9ba7c..08c709d 100755
--- a/flink-dist/src/main/flink-bin/bin/config.sh
+++ b/flink-dist/src/main/flink-bin/bin/config.sh
@@ -641,22 +641,29 @@ runBashJavaUtilsCmd() {
echo "$output"
}
-extractExecutionParams() {
- local execution_config=$1
+extractExecutionResults() {
+ local output="$1"
+ local expected_lines="$2"
local EXECUTION_PREFIX="BASH_JAVA_UTILS_EXEC_RESULT:"
+ local execution_results
- local num_lines=$(echo "$execution_config" | wc -l)
- if ! [[ ${num_lines} == 1 ]]; then
- echo "[ERROR] Unexpected result ($num_lines lines): $execution_config" 1>&2
- echo "[ERROR] extractExecutionParams only accepts exactly one line as the input" 1>&2
+ IFS=$'\n' execution_results=($(echo "${output}" | grep ${EXECUTION_PREFIX}))
+ if [[ ${#execution_results[@]} != ${expected_lines} ]]; then
+ echo "[ERROR] The execution results has unexpected number of lines, expected: ${expected_lines}, actual: ${#execution_results[@]}." 1>&2
+ echo "[ERROR] An execution result line is expected following the prefix '${EXECUTION_PREFIX}'" 1>&2
+ echo "$output" 1>&2
exit 1
fi
- if ! [[ ${execution_config} =~ ^${EXECUTION_PREFIX}.* ]]; then
- echo "[ERROR] Unexpected result: $execution_config" 1>&2
- echo "[ERROR] The output of BashJavaUtils is expected to be the execution result, following the prefix '${EXECUTION_PREFIX}'" 1>&2
- exit 1
- fi
+ for result in ${execution_results[@]}
+ do
+ echo ${result} | sed "s/$EXECUTION_PREFIX//"
+ done
+}
+
+extractLoggingOutputs() {
+ local output="$1"
+ local EXECUTION_PREFIX="BASH_JAVA_UTILS_EXEC_RESULT:"
- echo ${execution_config} | sed "s/$EXECUTION_PREFIX//"
+ echo "${output}" | grep -v ${EXECUTION_PREFIX}
}
diff --git a/flink-dist/src/main/flink-bin/bin/taskmanager.sh b/flink-dist/src/main/flink-bin/bin/taskmanager.sh
index 4b78a05..c6e3aab 100755
--- a/flink-dist/src/main/flink-bin/bin/taskmanager.sh
+++ b/flink-dist/src/main/flink-bin/bin/taskmanager.sh
@@ -50,24 +50,18 @@ if [[ $STARTSTOP == "start" ]] || [[ $STARTSTOP == "start-foreground" ]]; then
java_utils_output=$(runBashJavaUtilsCmd GET_TM_RESOURCE_PARAMS ${FLINK_CONF_DIR} $FLINK_BIN_DIR/bash-java-utils.jar:$(findFlinkDistJar) "${ARGS[@]}")
- num_lines=$(echo "${java_utils_output}" | wc -l)
- logging_output=$(echo "${java_utils_output}" | head -n $((${num_lines} - 2)))
- params_output=$(echo "${java_utils_output}" | tail -n 2)
+ logging_output=$(extractLoggingOutputs "${java_utils_output}")
+ params_output=$(extractExecutionResults "${java_utils_output}" 2)
- jvm_params=$(extractExecutionParams "$(echo "$params_output" | head -n 1)")
if [[ $? -ne 0 ]]; then
- echo "[ERROR] Could not get JVM parameters properly."
+ echo "[ERROR] Could not get JVM parameters and dynamic configurations properly."
exit 1
fi
- export JVM_ARGS="${JVM_ARGS} ${jvm_params}"
- IFS=$" "
+ jvm_params=$(echo "${params_output}" | head -n 1)
+ export JVM_ARGS="${JVM_ARGS} ${jvm_params}"
- dynamic_configs=$(extractExecutionParams "$(echo "$params_output" | tail -n 1)")
- if [[ $? -ne 0 ]]; then
- echo "[ERROR] Could not get dynamic configurations properly."
- exit 1
- fi
+ IFS=$" " dynamic_configs=$(echo "${params_output}" | tail -n 1)
ARGS+=("--configDir" "${FLINK_CONF_DIR}" ${dynamic_configs[@]})
export FLINK_INHERITED_LOGS="
diff --git a/flink-dist/src/test/bin/runBashJavaUtilsCmd.sh b/flink-dist/src/test/bin/runBashJavaUtilsCmd.sh
index 475d463..26a0f99 100755
--- a/flink-dist/src/test/bin/runBashJavaUtilsCmd.sh
+++ b/flink-dist/src/test/bin/runBashJavaUtilsCmd.sh
@@ -21,8 +21,8 @@
USAGE="Usage: runBashJavaUtilsCmd.sh <command> [dynamic args...]"
COMMAND=$1
-shift
-DYNAMIC_OPTS=$@
+EXPECTED_LINES=$2
+DYNAMIC_OPTS=${@:3}
if [[ -z "${COMMAND}" ]]; then
echo "$USAGE"
@@ -38,6 +38,5 @@ FLINK_DIST_JAR=`find $FLINK_TARGET_DIR -name 'flink-dist*.jar'`
. ${bin}/../../main/flink-bin/bin/config.sh > /dev/null
-output=$(runBashJavaUtilsCmd ${COMMAND} ${FLINK_CONF_DIR} "$FLINK_TARGET_DIR/bash-java-utils.jar:$FLINK_DIST_JAR}" $DYNAMIC_OPTS | tail -n 2)
-extractExecutionParams "$(echo "$output" | head -n 1)"
-extractExecutionParams "$(echo "$output" | tail -n 1)"
+output=$(runBashJavaUtilsCmd ${COMMAND} ${FLINK_CONF_DIR} "$FLINK_TARGET_DIR/bash-java-utils.jar:$FLINK_DIST_JAR}" $DYNAMIC_OPTS)
+extractExecutionResults "${output}" ${EXPECTED_LINES}
diff --git a/flink-dist/src/test/bin/runBashJavaUtilsCmd.sh b/flink-dist/src/test/bin/runExtractLoggingOutputs.sh
similarity index 73%
copy from flink-dist/src/test/bin/runBashJavaUtilsCmd.sh
copy to flink-dist/src/test/bin/runExtractLoggingOutputs.sh
index 475d463..d892901 100755
--- a/flink-dist/src/test/bin/runBashJavaUtilsCmd.sh
+++ b/flink-dist/src/test/bin/runExtractLoggingOutputs.sh
@@ -17,14 +17,11 @@
# limitations under the License.
################################################################################
-# Wrapper script to compare the TM heap size calculation of config.sh with Java
-USAGE="Usage: runBashJavaUtilsCmd.sh <command> [dynamic args...]"
+USAGE="Usage: runExtractLoggingOutputs.sh <input>"
-COMMAND=$1
-shift
-DYNAMIC_OPTS=$@
+INPUT="$1"
-if [[ -z "${COMMAND}" ]]; then
+if [[ -z "${INPUT}" ]]; then
echo "$USAGE"
exit 1
fi
@@ -38,6 +35,4 @@ FLINK_DIST_JAR=`find $FLINK_TARGET_DIR -name 'flink-dist*.jar'`
. ${bin}/../../main/flink-bin/bin/config.sh > /dev/null
-output=$(runBashJavaUtilsCmd ${COMMAND} ${FLINK_CONF_DIR} "$FLINK_TARGET_DIR/bash-java-utils.jar:$FLINK_DIST_JAR}" $DYNAMIC_OPTS | tail -n 2)
-extractExecutionParams "$(echo "$output" | head -n 1)"
-extractExecutionParams "$(echo "$output" | tail -n 1)"
+extractLoggingOutputs "${INPUT}"
diff --git a/flink-dist/src/test/java/org/apache/flink/dist/BashJavaUtilsITCase.java b/flink-dist/src/test/java/org/apache/flink/dist/BashJavaUtilsITCase.java
index f76b9c4..8aafb23 100644
--- a/flink-dist/src/test/java/org/apache/flink/dist/BashJavaUtilsITCase.java
+++ b/flink-dist/src/test/java/org/apache/flink/dist/BashJavaUtilsITCase.java
@@ -24,6 +24,7 @@ import org.apache.flink.runtime.util.BashJavaUtils;
import org.junit.Test;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
@@ -39,25 +40,53 @@ import static org.junit.Assert.assertThat;
public class BashJavaUtilsITCase extends JavaBashTestBase {
private static final String RUN_BASH_JAVA_UTILS_CMD_SCRIPT = "src/test/bin/runBashJavaUtilsCmd.sh";
+ private static final String RUN_EXTRACT_LOGGING_OUTPUTS_SCRIPT = "src/test/bin/runExtractLoggingOutputs.sh";
@Test
public void testGetTmResourceParamsConfigs() throws Exception {
- String[] commands = {RUN_BASH_JAVA_UTILS_CMD_SCRIPT, BashJavaUtils.Command.GET_TM_RESOURCE_PARAMS.toString()};
+ int expectedResultLines = 2;
+ String[] commands = {RUN_BASH_JAVA_UTILS_CMD_SCRIPT, BashJavaUtils.Command.GET_TM_RESOURCE_PARAMS.toString(), String.valueOf(expectedResultLines)};
List<String> lines = Arrays.asList(executeScript(commands).split(System.lineSeparator()));
- assertThat(lines.size(), is(2));
+ assertThat(lines.size(), is(expectedResultLines));
ConfigurationUtils.parseJvmArgString(lines.get(0));
ConfigurationUtils.parseTmResourceDynamicConfigs(lines.get(1));
}
@Test
public void testConfigOverwrittenByDynamicOpts() throws Exception {
+ int expectedResultLines = 2;
double cpuCores = 39.0;
- String[] commands = {RUN_BASH_JAVA_UTILS_CMD_SCRIPT, BashJavaUtils.Command.GET_TM_RESOURCE_PARAMS.toString(), "-D" + TaskManagerOptions.CPU_CORES.key() + "=" + cpuCores};
+ String[] commands = {
+ RUN_BASH_JAVA_UTILS_CMD_SCRIPT,
+ BashJavaUtils.Command.GET_TM_RESOURCE_PARAMS.toString(),
+ String.valueOf(expectedResultLines),
+ "-D" + TaskManagerOptions.CPU_CORES.key() + "=" + cpuCores};
List<String> lines = Arrays.asList(executeScript(commands).split(System.lineSeparator()));
- assertThat(lines.size(), is(2));
+ assertThat(lines.size(), is(expectedResultLines));
Map<String, String> configs = ConfigurationUtils.parseTmResourceDynamicConfigs(lines.get(1));
assertThat(Double.valueOf(configs.get(TaskManagerOptions.CPU_CORES.key())), is(cpuCores));
}
+
+ @Test
+ public void testExtractLoggingOutputs() throws Exception {
+ StringBuilder input = new StringBuilder();
+ List<String> expectedOutput = new ArrayList<>();
+
+ for (int i = 0; i < 5; ++i) {
+ String line = "BashJavaUtils output line " + i + " `~!@#$%^&*()-_=+;:,.'\"\\\t/?";
+ if (i % 2 == 0) {
+ expectedOutput.add(line);
+ } else {
+ line = BashJavaUtils.EXECUTION_PREFIX + line;
+ }
+ input.append(line + "\n");
+ }
+
+ String[] commands = {RUN_EXTRACT_LOGGING_OUTPUTS_SCRIPT, input.toString()};
+ List<String> actualOutput = Arrays.asList(executeScript(commands).split(System.lineSeparator()));
+
+ assertThat(actualOutput, is(expectedOutput));
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/BashJavaUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/BashJavaUtils.java
index 14d24f5..e00d667 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/util/BashJavaUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/BashJavaUtils.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.util;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.clusterframework.TaskExecutorProcessSpec;
@@ -34,7 +35,8 @@ import static org.apache.flink.util.Preconditions.checkArgument;
*/
public class BashJavaUtils {
- private static final String EXECUTION_PREFIX = "BASH_JAVA_UTILS_EXEC_RESULT:";
+ @VisibleForTesting
+ public static final String EXECUTION_PREFIX = "BASH_JAVA_UTILS_EXEC_RESULT:";
public static void main(String[] args) throws Exception {
checkArgument(args.length > 0, "Command not specified.");
[flink] 12/17: [FLINK-16745][coordination] Remove unused container
cutoff
Posted by az...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
azagrebin pushed a commit to branch flip116
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 46e0dc3aaa34f921162feb38b26f2ca29ce7204b
Author: Andrey Zagrebin <az...@apache.org>
AuthorDate: Wed Apr 8 19:43:33 2020 +0300
[FLINK-16745][coordination] Remove unused container cutoff
---
.../generated/resource_manager_configuration.html | 12 ----
docs/ops/memory/mem_migration.md | 5 +-
docs/ops/memory/mem_migration.zh.md | 5 +-
.../flink/configuration/ConfigConstants.java | 12 ++--
.../configuration/ResourceManagerOptions.java | 20 -------
.../test-scripts/common_yarn_docker.sh | 1 -
.../test-scripts/test_kubernetes_session.sh | 1 -
.../runtime/clusterframework/BootstrapTools.java | 28 ---------
.../clusterframework/BootstrapToolsTest.java | 53 -----------------
.../StandaloneResourceManagerFactoryTest.java | 68 ----------------------
.../flink/yarn/YARNHighAvailabilityITCase.java | 4 --
.../yarn/YARNSessionCapacitySchedulerITCase.java | 5 --
.../apache/flink/yarn/YarnConfigurationITCase.java | 4 --
.../flink/yarn/YarnClusterDescriptorTest.java | 3 -
.../apache/flink/yarn/YarnResourceManagerTest.java | 2 -
15 files changed, 9 insertions(+), 214 deletions(-)
diff --git a/docs/_includes/generated/resource_manager_configuration.html b/docs/_includes/generated/resource_manager_configuration.html
index c1955f8..cb79c38 100644
--- a/docs/_includes/generated/resource_manager_configuration.html
+++ b/docs/_includes/generated/resource_manager_configuration.html
@@ -9,18 +9,6 @@
</thead>
<tbody>
<tr>
- <td><h5>containerized.heap-cutoff-min</h5></td>
- <td style="word-wrap: break-word;">600</td>
- <td>Integer</td>
- <td>Minimum amount of heap memory to remove in Job Master containers, as a safety margin.</td>
- </tr>
- <tr>
- <td><h5>containerized.heap-cutoff-ratio</h5></td>
- <td style="word-wrap: break-word;">0.25</td>
- <td>Float</td>
- <td>Percentage of heap space to remove from Job Master containers (YARN / Mesos / Kubernetes), to compensate for other JVM memory usage.</td>
- </tr>
- <tr>
<td><h5>resourcemanager.job.timeout</h5></td>
<td style="word-wrap: break-word;">"5 minutes"</td>
<td>String</td>
diff --git a/docs/ops/memory/mem_migration.md b/docs/ops/memory/mem_migration.md
index e191940..eb2db4f 100644
--- a/docs/ops/memory/mem_migration.md
+++ b/docs/ops/memory/mem_migration.md
@@ -125,9 +125,8 @@ Although, the network memory configuration has not changed too much it is recomm
It can change if other memory components have new sizes, e.g. the total memory which the network can be a fraction of.
See also [new detailed memory model](mem_detail.html).
-The container cut-off configuration options, [`containerized.heap-cutoff-ratio`](config.html#containerized-heap-cutoff-ratio)
-and [`containerized.heap-cutoff-min`](config.html#containerized-heap-cutoff-min), have no effect for task manager processes anymore
-but they still have the same semantics for the job manager process. See also [how to migrate container cut-off](#container-cut-off-memory).
+The container cut-off configuration options, `containerized.heap-cutoff-ratio` and `containerized.heap-cutoff-min`,
+have no effect anymore. See also [how to migrate container cut-off](#container-cut-off-memory).
## Total Memory (Previously Heap Memory)
diff --git a/docs/ops/memory/mem_migration.zh.md b/docs/ops/memory/mem_migration.zh.md
index d51683a..72d2b01 100644
--- a/docs/ops/memory/mem_migration.zh.md
+++ b/docs/ops/memory/mem_migration.zh.md
@@ -121,10 +121,7 @@ Flink 自带的[默认 flink-conf.yaml](#flink-confyaml-中的默认配置) 文
网络内存的大小可能会受到其他内存部分大小变化的影响，例如总内存变化时，根据占比计算出的网络内存也可能发生变化。
请参考[内存模型详解](mem_detail.html)。
-容器切除（Cut-Off）内存相关的配置参数（[`containerized.heap-cutoff-ratio`](config.html#containerized-heap-cutoff-ratio)
-和 [`containerized.heap-cutoff-min`](config.html#containerized-heap-cutoff-min)）将不再对 TaskExecutor 进程生效。
-对于 JobManager 进程，它们仍具有与此前相同的语义。
-请参考[如何升级容器切除内存](#容器切除cut-off内存)。
+容器切除（Cut-Off）内存相关的配置参数（`containerized.heap-cutoff-ratio` 和 `containerized.heap-cutoff-min`）将不再对进程生效。
## 总内存（原堆内存）
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
index 16fccb1..f1e2de8 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
@@ -392,14 +392,14 @@ public final class ConfigConstants {
/**
* Percentage of heap space to remove from containers (YARN / Mesos / Kubernetes), to compensate
* for other JVM memory usage.
- * @deprecated Use {@link ResourceManagerOptions#CONTAINERIZED_HEAP_CUTOFF_RATIO} instead.
+ * @deprecated Not used anymore, but remain here until Flink 2.0
*/
@Deprecated
public static final String CONTAINERIZED_HEAP_CUTOFF_RATIO = "containerized.heap-cutoff-ratio";
/**
* Minimum amount of heap memory to remove in containers, as a safety margin.
- * @deprecated Use {@link ResourceManagerOptions#CONTAINERIZED_HEAP_CUTOFF_MIN} instead.
+ * @deprecated Not used anymore, but remain here until Flink 2.0
*/
@Deprecated
public static final String CONTAINERIZED_HEAP_CUTOFF_MIN = "containerized.heap-cutoff-min";
@@ -434,14 +434,14 @@ public final class ConfigConstants {
/**
* Percentage of heap space to remove from containers started by YARN.
- * @deprecated in favor of {@code #CONTAINERIZED_HEAP_CUTOFF_RATIO}
+ * @deprecated Not used anymore, but remain here until Flink 2.0
*/
@Deprecated
public static final String YARN_HEAP_CUTOFF_RATIO = "yarn.heap-cutoff-ratio";
/**
* Minimum amount of memory to remove from the heap space as a safety margin.
- * @deprecated in favor of {@code #CONTAINERIZED_HEAP_CUTOFF_MIN}
+ * @deprecated Not used anymore, but remain here until Flink 2.0
*/
@Deprecated
public static final String YARN_HEAP_CUTOFF_MIN = "yarn.heap-cutoff-min";
@@ -1526,7 +1526,7 @@ public final class ConfigConstants {
/**
* Minimum amount of memory to subtract from the process memory to get the TaskManager
* heap size. We came up with these values experimentally.
- * @deprecated Use {@link ResourceManagerOptions#CONTAINERIZED_HEAP_CUTOFF_MIN} instead.
+ * @deprecated Not used anymore, but remain here until Flink 2.0
*/
@Deprecated
public static final int DEFAULT_YARN_HEAP_CUTOFF = 600;
@@ -1534,7 +1534,7 @@ public final class ConfigConstants {
/**
* Relative amount of memory to subtract from Java process memory to get the TaskManager
* heap size.
- * @deprecated Use {@link ResourceManagerOptions#CONTAINERIZED_HEAP_CUTOFF_RATIO} instead.
+ * @deprecated Not used anymore, but remain here until Flink 2.0
*/
@Deprecated
public static final float DEFAULT_YARN_HEAP_CUTOFF_RATIO = 0.25f;
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ResourceManagerOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/ResourceManagerOptions.java
index fde942b..4af839f 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/ResourceManagerOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/ResourceManagerOptions.java
@@ -57,26 +57,6 @@ public class ResourceManagerOptions {
" Its not possible to use this configuration key to define port ranges.");
/**
- * Percentage of heap space to remove from Job Master containers (YARN / Mesos/ Kubernetes), to compensate
- * for other JVM memory usage.
- */
- public static final ConfigOption<Float> CONTAINERIZED_HEAP_CUTOFF_RATIO = ConfigOptions
- .key("containerized.heap-cutoff-ratio")
- .defaultValue(0.25f)
- .withDeprecatedKeys("yarn.heap-cutoff-ratio")
- .withDescription("Percentage of heap space to remove from Job Master containers (YARN / Mesos / Kubernetes), " +
- "to compensate for other JVM memory usage.");
-
- /**
- * Minimum amount of heap memory to remove in Job Master containers, as a safety margin.
- */
- public static final ConfigOption<Integer> CONTAINERIZED_HEAP_CUTOFF_MIN = ConfigOptions
- .key("containerized.heap-cutoff-min")
- .defaultValue(600)
- .withDeprecatedKeys("yarn.heap-cutoff-min")
- .withDescription("Minimum amount of heap memory to remove in Job Master containers, as a safety margin.");
-
- /**
* The timeout for a slot request to be discarded, in milliseconds.
* @deprecated Use {@link JobManagerOptions#SLOT_REQUEST_TIMEOUT}.
*/
diff --git a/flink-end-to-end-tests/test-scripts/common_yarn_docker.sh b/flink-end-to-end-tests/test-scripts/common_yarn_docker.sh
index bcdff3e..34d6ef4 100755
--- a/flink-end-to-end-tests/test-scripts/common_yarn_docker.sh
+++ b/flink-end-to-end-tests/test-scripts/common_yarn_docker.sh
@@ -123,7 +123,6 @@ function start_hadoop_cluster_and_prepare_flink() {
security.kerberos.login.keytab: /home/hadoop-user/hadoop-user.keytab
security.kerberos.login.principal: hadoop-user
slot.request.timeout: 120000
-containerized.heap-cutoff-min: 100
END
)
docker exec master bash -c "echo \"$FLINK_CONFIG\" > /home/hadoop-user/$FLINK_DIRNAME/conf/flink-conf.yaml"
diff --git a/flink-end-to-end-tests/test-scripts/test_kubernetes_session.sh b/flink-end-to-end-tests/test-scripts/test_kubernetes_session.sh
index 1e450ec..b45e1c3 100755
--- a/flink-end-to-end-tests/test-scripts/test_kubernetes_session.sh
+++ b/flink-end-to-end-tests/test-scripts/test_kubernetes_session.sh
@@ -65,7 +65,6 @@ mkdir -p "$(dirname $LOCAL_OUTPUT_PATH)"
"$FLINK_DIR"/bin/kubernetes-session.sh -Dkubernetes.cluster-id=${CLUSTER_ID} \
-Dkubernetes.container.image=${FLINK_IMAGE_NAME} \
-Djobmanager.memory.process.size=768m \
- -Dcontainerized.heap-cutoff-min=100 \
-Dkubernetes.jobmanager.cpu=0.5 \
-Dkubernetes.taskmanager.cpu=0.5 \
-Dkubernetes.container-start-command-template="%java% %classpath% %jvmmem% %jvmopts% %logging% %class% %args%" \
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/BootstrapTools.java b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/BootstrapTools.java
index 741edd4..4c7db29 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/BootstrapTools.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/BootstrapTools.java
@@ -25,7 +25,6 @@ import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.CoreOptions;
-import org.apache.flink.configuration.ResourceManagerOptions;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.entrypoint.parser.CommandLineOptions;
import org.apache.flink.runtime.util.config.memory.ProcessMemoryUtils;
@@ -687,31 +686,4 @@ public class BootstrapTools {
public static String escapeWithDoubleQuote(String value) {
return "\"" + WINDOWS_DOUBLE_QUOTE_ESCAPER.escape(value) + "\"";
}
-
- /**
- * Calculate heap size after cut-off. The heap size after cut-off will be used to set -Xms and -Xmx for jobmanager
- * start command.
- */
- public static int calculateHeapSize(int memory, Configuration conf) {
-
- final float memoryCutoffRatio = conf.getFloat(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_RATIO);
- final int minCutoff = conf.getInteger(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_MIN);
-
- if (memoryCutoffRatio > 1 || memoryCutoffRatio < 0) {
- throw new IllegalArgumentException("The configuration value '"
- + ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_RATIO.key()
- + "' must be between 0 and 1. Value given=" + memoryCutoffRatio);
- }
- if (minCutoff > memory) {
- throw new IllegalArgumentException("The configuration value '"
- + ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_MIN.key()
- + "' is higher (" + minCutoff + ") than the requested amount of memory " + memory);
- }
-
- int heapLimit = (int) ((float) memory * memoryCutoffRatio);
- if (heapLimit < minCutoff) {
- heapLimit = minCutoff;
- }
- return memory - heapLimit;
- }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/BootstrapToolsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/BootstrapToolsTest.java
index cba22c5..909ba01 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/BootstrapToolsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/BootstrapToolsTest.java
@@ -27,7 +27,6 @@ import org.apache.flink.configuration.ConfigurationUtils;
import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.configuration.MemorySize;
-import org.apache.flink.configuration.ResourceManagerOptions;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.concurrent.FutureUtils;
import org.apache.flink.util.ExceptionUtils;
@@ -476,58 +475,6 @@ public class BootstrapToolsTest extends TestLogger {
}
@Test
- public void testHeapCutoff() {
- Configuration conf = new Configuration();
- conf.setFloat(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_RATIO, 0.15F);
- conf.setInteger(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_MIN, 384);
-
- Assert.assertEquals(616, BootstrapTools.calculateHeapSize(1000, conf));
- Assert.assertEquals(8500, BootstrapTools.calculateHeapSize(10000, conf));
-
- // test different configuration
- Assert.assertEquals(3400, BootstrapTools.calculateHeapSize(4000, conf));
-
- conf.setString(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_MIN.key(), "1000");
- conf.setString(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_RATIO.key(), "0.1");
- Assert.assertEquals(3000, BootstrapTools.calculateHeapSize(4000, conf));
-
- conf.setString(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_RATIO.key(), "0.5");
- Assert.assertEquals(2000, BootstrapTools.calculateHeapSize(4000, conf));
-
- conf.setString(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_RATIO.key(), "1");
- Assert.assertEquals(0, BootstrapTools.calculateHeapSize(4000, conf));
-
- // test also deprecated keys
- conf = new Configuration();
- conf.setDouble(ConfigConstants.YARN_HEAP_CUTOFF_RATIO, 0.15);
- conf.setInteger(ConfigConstants.YARN_HEAP_CUTOFF_MIN, 384);
-
- Assert.assertEquals(616, BootstrapTools.calculateHeapSize(1000, conf));
- Assert.assertEquals(8500, BootstrapTools.calculateHeapSize(10000, conf));
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void illegalArgument() {
- final Configuration conf = new Configuration();
- conf.setString(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_RATIO.key(), "1.1");
- BootstrapTools.calculateHeapSize(4000, conf);
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void illegalArgumentNegative() {
- final Configuration conf = new Configuration();
- conf.setString(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_RATIO.key(), "-0.01");
- BootstrapTools.calculateHeapSize(4000, conf);
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void tooMuchCutoff() {
- final Configuration conf = new Configuration();
- conf.setString(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_RATIO.key(), "6000");
- BootstrapTools.calculateHeapSize(4000, conf);
- }
-
- @Test
public void testGetEnvironmentVariables() {
Configuration testConf = new Configuration();
testConf.setString("containerized.master.env.LD_LIBRARY_PATH", "/usr/lib/native");
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManagerFactoryTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManagerFactoryTest.java
deleted file mode 100644
index 46aceb2..0000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManagerFactoryTest.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.resourcemanager;
-
-import org.apache.flink.api.common.time.Time;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.MemorySize;
-import org.apache.flink.configuration.ResourceManagerOptions;
-import org.apache.flink.configuration.TaskManagerOptions;
-import org.apache.flink.runtime.clusterframework.types.ResourceID;
-import org.apache.flink.runtime.entrypoint.ClusterInformation;
-import org.apache.flink.runtime.heartbeat.TestingHeartbeatServices;
-import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServices;
-import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
-import org.apache.flink.runtime.rpc.RpcUtils;
-import org.apache.flink.runtime.rpc.TestingRpcService;
-import org.apache.flink.runtime.util.TestingFatalErrorHandler;
-import org.apache.flink.util.TestLogger;
-
-import org.junit.Test;
-
-/**
- * Tests for the {@link StandaloneResourceManagerFactory}.
- */
-public class StandaloneResourceManagerFactoryTest extends TestLogger {
-
- @Test
- public void createResourceManager_WithLessMemoryThanContainerizedHeapCutoffMin_ShouldSucceed() throws Exception {
- final StandaloneResourceManagerFactory resourceManagerFactory = StandaloneResourceManagerFactory.INSTANCE;
-
- final TestingRpcService rpcService = new TestingRpcService();
- try {
- final Configuration configuration = new Configuration();
- configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, new MemorySize(128 * 1024 * 1024));
- configuration.setInteger(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_MIN, 600);
-
- final ResourceManager<ResourceID> ignored = resourceManagerFactory.createResourceManager(
- configuration,
- ResourceID.generate(),
- rpcService,
- new TestingHighAvailabilityServices(),
- new TestingHeartbeatServices(),
- new TestingFatalErrorHandler(),
- new ClusterInformation("foobar", 1234),
- null,
- UnregisteredMetricGroups.createUnregisteredResourceManagerMetricGroup());
- } finally {
- RpcUtils.terminateRpcService(rpcService, Time.seconds(10L));
- }
- }
-
-}
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
index b4ec5ea..a8b799a 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNHighAvailabilityITCase.java
@@ -28,7 +28,6 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.HighAvailabilityOptions;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.MemorySize;
-import org.apache.flink.configuration.ResourceManagerOptions;
import org.apache.flink.configuration.RestartStrategyOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.execution.ExecutionState;
@@ -285,9 +284,6 @@ public class YARNHighAvailabilityITCase extends YarnTestBase {
flinkConfiguration.setString(RestartStrategyOptions.RESTART_STRATEGY, "fixed-delay");
flinkConfiguration.setInteger(RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS, Integer.MAX_VALUE);
- final int minMemory = 100;
- flinkConfiguration.setInteger(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_MIN, minMemory);
-
return createYarnClusterDescriptor(flinkConfiguration);
}
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
index cd310b2..c8b6280 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
@@ -24,7 +24,6 @@ import org.apache.flink.client.cli.CliFrontend;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.configuration.JobManagerOptions;
-import org.apache.flink.configuration.ResourceManagerOptions;
import org.apache.flink.runtime.rest.RestClient;
import org.apache.flink.runtime.rest.RestClientConfiguration;
import org.apache.flink.runtime.rest.handler.legacy.messages.ClusterOverviewWithVersion;
@@ -498,9 +497,6 @@ public class YARNSessionCapacitySchedulerITCase extends YarnTestBase {
"-yt", flinkLibFolder.getAbsolutePath(),
"-yt", flinkShadedHadoopDir.getAbsolutePath(),
"-yjm", "768m",
- // test if the cutoff is passed correctly (only useful when larger than the value
- // of containerized.heap-cutoff-min (default: 600MB)
- "-yD", ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_RATIO.key() + "=0.7",
"-yD", YarnConfigOptions.APPLICATION_TAGS.key() + "=test-tag",
"-ytm", "1024m",
"-ys", "2", // test requesting slots from YARN.
@@ -580,7 +576,6 @@ public class YARNSessionCapacitySchedulerITCase extends YarnTestBase {
});
Assert.assertNotNull("Unable to locate JobManager log", jobmanagerLog);
content = FileUtils.readFileToString(jobmanagerLog);
- // TM was started with 1024 but we cut off 70% (NOT THE DEFAULT VALUE)
String expected = "Starting TaskManagers";
Assert.assertTrue("Expected string '" + expected + "' not found in JobManager log: '" + jobmanagerLog + "'",
content.contains(expected));
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java
index ee538d4..900abb6 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java
@@ -25,7 +25,6 @@ import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.PackagedProgram;
import org.apache.flink.client.program.PackagedProgramUtils;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.ResourceManagerOptions;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.TaskManagerOptions;
@@ -85,9 +84,6 @@ public class YarnConfigurationITCase extends YarnTestBase {
final YarnClient yarnClient = getYarnClient();
final Configuration configuration = new Configuration(flinkConfiguration);
- // disable heap cutoff min
- configuration.setInteger(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_MIN, 0);
-
final int slotsPerTaskManager = 3;
configuration.set(TaskManagerOptions.NUM_TASK_SLOTS, slotsPerTaskManager);
final int masterMemory = 768;
diff --git a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java
index 09d2133..661e0d3 100644
--- a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java
+++ b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnClusterDescriptorTest.java
@@ -24,7 +24,6 @@ import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.configuration.IllegalConfigurationException;
-import org.apache.flink.configuration.ResourceManagerOptions;
import org.apache.flink.core.testutils.CommonTestUtils;
import org.apache.flink.runtime.jobmanager.JobManagerProcessSpec;
import org.apache.flink.runtime.util.config.memory.ProcessMemoryUtils;
@@ -101,7 +100,6 @@ public class YarnClusterDescriptorTest extends TestLogger {
@Test
public void testFailIfTaskSlotsHigherThanMaxVcores() throws ClusterDeploymentException {
final Configuration flinkConfiguration = new Configuration();
- flinkConfiguration.setInteger(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_MIN, 0);
YarnClusterDescriptor clusterDescriptor = createYarnClusterDescriptor(flinkConfiguration);
@@ -130,7 +128,6 @@ public class YarnClusterDescriptorTest extends TestLogger {
Configuration configuration = new Configuration();
// overwrite vcores in config
configuration.setInteger(YarnConfigOptions.VCORES, Integer.MAX_VALUE);
- configuration.setInteger(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_MIN, 0);
YarnClusterDescriptor clusterDescriptor = createYarnClusterDescriptor(configuration);
diff --git a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnResourceManagerTest.java b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnResourceManagerTest.java
index 122029c..7de870f 100755
--- a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnResourceManagerTest.java
+++ b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnResourceManagerTest.java
@@ -23,7 +23,6 @@ import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.configuration.MemorySize;
-import org.apache.flink.configuration.ResourceManagerOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.clusterframework.ApplicationStatus;
import org.apache.flink.runtime.clusterframework.BootstrapTools;
@@ -148,7 +147,6 @@ public class YarnResourceManagerTest extends TestLogger {
testingFatalErrorHandler = new TestingFatalErrorHandler();
flinkConfig = new Configuration();
- flinkConfig.setInteger(ResourceManagerOptions.CONTAINERIZED_HEAP_CUTOFF_MIN, 100);
flinkConfig.set(TaskManagerOptions.TOTAL_FLINK_MEMORY, MemorySize.parse("1g"));
File root = folder.getRoot();