You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by bh...@apache.org on 2020/06/03 01:51:47 UTC
[samza] branch master updated: SAMZA-2538: Update build.gradle from
using config-path to job.config.loader.properties.path
This is an automated email from the ASF dual-hosted git repository.
bharathkk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/samza.git
The following commit(s) were added to refs/heads/master by this push:
new b3d5dc3 SAMZA-2538: Update build.gradle from using config-path to job.config.loader.properties.path
b3d5dc3 is described below
commit b3d5dc3937830898d9213acf9156519fb574492c
Author: Ke Wu <kw...@linkedin.com>
AuthorDate: Tue Jun 2 18:51:19 2020 -0700
SAMZA-2538: Update build.gradle from using config-path to job.config.loader.properties.path
Symptom: Gradle tasks in samza-shell (checkpointTool, kvPerformanceTest, runJob) does not work.
Cause: Gradle tasks are not being updated to be compatible with Samza 1.5.
Changes: Translate configPath property to job.config.loader.factory and job.config.loader.properties.path instead of config-path.
Tests:
1. ./gradlew samza-shell:runJob -PconfigPath=/deploy/samza/config/wikipedia-feed.properties
2. ./gradlew samza-shell:kvPerformanceTest -PconfigPath=$PWD/samza-test/src/main/config/perf/kv-perf.properties
API Changes: None
Upgrade Instructions: None
Usage Instructions: None
Author: Ke Wu <kw...@linkedin.com>
Reviewers: mynameborat <bh...@apache.org>
Closes #1371 from kw2542/SAMZA-2538
---
README.md | 6 +++---
build.gradle | 18 ++++++++++++------
.../documentation/versioned/container/checkpointing.md | 6 +++---
.../versioned/container/coordinator-stream.md | 2 +-
.../versioned/container/state-management.md | 4 ++--
gradle/wrapper/gradle-wrapper.properties | 1 +
.../org/apache/samza/checkpoint/CheckpointTool.scala | 9 ++++-----
7 files changed, 26 insertions(+), 20 deletions(-)
diff --git a/README.md b/README.md
index ec1cd94..e5c91a4 100644
--- a/README.md
+++ b/README.md
@@ -62,15 +62,15 @@ To run standalone integration tests:
To run a job (defined in a properties file):
- ./gradlew samza-shell:runJob -PconfigPath=file:///path/to/job/config.properties
+ ./gradlew samza-shell:runJob -PconfigPath=/path/to/job/config.properties
To inspect a job's latest checkpoint:
- ./gradlew samza-shell:checkpointTool -PconfigPath=file:///path/to/job/config.properties
+ ./gradlew samza-shell:checkpointTool -PconfigPath=/path/to/job/config.properties
To modify a job's checkpoint (assumes that the job is not currently running), give it a file with the new offset for each partition, in the format `systems.<system>.streams.<topic>.partitions.<partition>=<offset>`:
- ./gradlew samza-shell:checkpointTool -PconfigPath=file:///path/to/job/config.properties \
+ ./gradlew samza-shell:checkpointTool -PconfigPath=/path/to/job/config.properties \
-PnewOffsets=file:///path/to/new/offsets.properties
### Developers
diff --git a/build.gradle b/build.gradle
index 243fce8..41ff784 100644
--- a/build.gradle
+++ b/build.gradle
@@ -575,33 +575,39 @@ project(":samza-shell") {
}
// Usage: ./gradlew samza-shell:runJob \
- // -PconfigPath=file:///path/to/job/config.properties
+ // -PconfigPath=/path/to/job/config.properties
task runJob(type:JavaExec) {
description 'To run a job (defined in a properties file)'
main = 'org.apache.samza.job.JobRunner'
classpath = configurations.gradleShell
- if (project.hasProperty('configPath')) args += ['--config-path', configPath]
+ if (project.hasProperty('configPath')) args += [
+ '--config', 'job.config.loader.factory=org.apache.samza.config.loaders.PropertiesConfigLoaderFactory',
+ '--config', 'job.config.loader.properties.path=' + configPath]
jvmArgs = ["-Dlog4j.configurationFile=file:src/main/resources/log4j2-console.xml"]
}
// Usage: ./gradlew samza-shell:checkpointTool \
- // -PconfigPath=file:///path/to/job/config.properties -PnewOffsets=file:///path/to/new/offsets.properties
+ // -PconfigPath=/path/to/job/config.properties -PnewOffsets=/path/to/new/offsets.properties
task checkpointTool(type:JavaExec) {
description 'Command-line tool to inspect and manipulate the job’s checkpoint'
main = 'org.apache.samza.checkpoint.CheckpointTool'
classpath = configurations.gradleShell
- if (project.hasProperty('configPath')) args += ['--config-path', configPath]
+ if (project.hasProperty('configPath')) args += [
+ '--config', 'job.config.loader.factory=org.apache.samza.config.loaders.PropertiesConfigLoaderFactory',
+ '--config', 'job.config.loader.properties.path=' + configPath]
if (project.hasProperty('newOffsets')) args += ['--new-offsets', newOffsets]
jvmArgs = ["-Dlog4j.configurationFile=file:src/main/resources/log4j2-console.xml"]
}
// Usage: ./gradlew samza-shell:kvPerformanceTest
- // -PconfigPath=file:///path/to/job/config.properties
+ // -PconfigPath=/path/to/job/config.properties
task kvPerformanceTest(type:JavaExec) {
description 'Command-line tool to run key-value performance tests'
main = 'org.apache.samza.test.performance.TestKeyValuePerformance'
classpath = configurations.gradleShell
- if (project.hasProperty('configPath')) args += ['--config-path', configPath]
+ if (project.hasProperty('configPath')) args += [
+ '--config', 'job.config.loader.factory=org.apache.samza.config.loaders.PropertiesConfigLoaderFactory',
+ '--config', 'job.config.loader.properties.path=' + configPath]
jvmArgs = ["-Dlog4j.configurationFile=file:src/main/resources/log4j2-console.xml"]
}
}
diff --git a/docs/learn/documentation/versioned/container/checkpointing.md b/docs/learn/documentation/versioned/container/checkpointing.md
index 9fb7e6d..a829b1f 100644
--- a/docs/learn/documentation/versioned/container/checkpointing.md
+++ b/docs/learn/documentation/versioned/container/checkpointing.md
@@ -108,15 +108,15 @@ To inspect a job's latest checkpoint, you need to specify your job's config file
{% highlight bash %}
samza-example/target/bin/checkpoint-tool.sh \
- --config-path=file:///path/to/job/config.properties
+ --config-path=/path/to/job/config.properties
{% endhighlight %}
This command prints out the latest checkpoint in a properties file format. You can save the output to a file, and edit it as you wish. For example, to jump back to the oldest possible point in time, you can set all the offsets to 0. Then you can feed that properties file back into checkpoint-tool.sh and save the modified checkpoint:
{% highlight bash %}
samza-example/target/bin/checkpoint-tool.sh \
- --config-path=file:///path/to/job/config.properties \
- --new-offsets=file:///path/to/new/offsets.properties
+ --config-path=/path/to/job/config.properties \
+ --new-offsets=/path/to/new/offsets.properties
{% endhighlight %}
Note that Samza only reads checkpoints on container startup. In order for your checkpoint change to take effect, you need to first stop the job, then save the modified offsets, and then start the job again. If you write a checkpoint while the job is running, it will most likely have no effect.
diff --git a/docs/learn/documentation/versioned/container/coordinator-stream.md b/docs/learn/documentation/versioned/container/coordinator-stream.md
index 6b14960..8d528d6 100644
--- a/docs/learn/documentation/versioned/container/coordinator-stream.md
+++ b/docs/learn/documentation/versioned/container/coordinator-stream.md
@@ -114,7 +114,7 @@ The coordinator stream messages that are currently supported are listed below:
Samza provides a command line tool to write Job Configuration messages to the coordinator stream. The tool can be used as follows:
{% highlight bash %}
samza-example/target/bin/run-coordinator-stream-writer.sh \
- --config-path=file:///path/to/job/config.properties \
+ --config-path=/path/to/job/config.properties \
--type set-config \
--key job.container.count \
--value 8
diff --git a/docs/learn/documentation/versioned/container/state-management.md b/docs/learn/documentation/versioned/container/state-management.md
index 86b0d44..4371541 100644
--- a/docs/learn/documentation/versioned/container/state-management.md
+++ b/docs/learn/documentation/versioned/container/state-management.md
@@ -192,7 +192,7 @@ Currently Samza provides a state storage tool which can recover the state store
{% highlight bash %}
samza-example/target/bin/state-storage-tool.sh \
- --config-path=file:///path/to/job/config.properties \
+ --config-path=/path/to/job/config.properties \
--path=directory/to/put/state/stores
{% endhighlight %}
@@ -202,7 +202,7 @@ Samza also provides a tool to read the value from a running job's RocksDB.
{% highlight bash %}
samza-example/target/bin/read-rocksdb-tool.sh \
- --config-path=file:///path/to/job/config.properties \
+ --config-path=/path/to/job/config.properties \
--db-path=/tmp/nm-local-dir/state/test-state/Partition_0 \
--db-name=test-state \
--string-key=a,b,c
diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties
index 44e7c4d..83639a3 100644
--- a/gradle/wrapper/gradle-wrapper.properties
+++ b/gradle/wrapper/gradle-wrapper.properties
@@ -1,3 +1,4 @@
+#Mon Jun 01 15:50:38 PDT 2020
distributionBase=GRADLE_USER_HOME
distributionPath=wrapper/dists
distributionUrl=https\://services.gradle.org/distributions/gradle-5.2.1-bin.zip
diff --git a/samza-core/src/main/scala/org/apache/samza/checkpoint/CheckpointTool.scala b/samza-core/src/main/scala/org/apache/samza/checkpoint/CheckpointTool.scala
index 4db807e..7936c0a 100644
--- a/samza-core/src/main/scala/org/apache/samza/checkpoint/CheckpointTool.scala
+++ b/samza-core/src/main/scala/org/apache/samza/checkpoint/CheckpointTool.scala
@@ -20,7 +20,6 @@
package org.apache.samza.checkpoint
import java.io.FileInputStream
-import java.net.URI
import java.util
import java.util.Properties
import java.util.regex.Pattern
@@ -82,12 +81,12 @@ object CheckpointTool {
type TaskNameToCheckpointMap = Map[TaskName, Map[SystemStreamPartition, String]]
class CheckpointToolCommandLine extends CommandLine with Logging {
- val newOffsetsOpt: ArgumentAcceptingOptionSpec[URI] =
- parser.accepts("new-offsets", "URI of file (e.g. file:///some/local/path.properties) " +
+ val newOffsetsOpt: ArgumentAcceptingOptionSpec[String] =
+ parser.accepts("new-offsets", "Location of file (e.g. /some/local/path.properties) " +
"containing offsets to write to the job's checkpoint topic. " +
"If not given, this tool prints out the current offsets.")
.withRequiredArg
- .ofType(classOf[URI])
+ .ofType(classOf[String])
.describedAs("path")
var newOffsets: TaskNameToCheckpointMap = _
@@ -121,7 +120,7 @@ object CheckpointTool {
override def loadConfig(options: OptionSet): Config = {
val config = super.loadConfig(options)
if (options.has(newOffsetsOpt)) {
- val newOffsetsInputStream = new FileInputStream(options.valueOf(newOffsetsOpt).getPath)
+ val newOffsetsInputStream = new FileInputStream(options.valueOf(newOffsetsOpt))
val properties = new Properties()
properties.load(newOffsetsInputStream)