You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by xi...@apache.org on 2017/05/22 16:36:41 UTC
samza git commit: SAMZA-1291: StandAlone config
Repository: samza
Updated Branches:
refs/heads/master f41fd2545 -> 1e9ae7395
SAMZA-1291: StandAlone config
Author: Boris Shkolnik <bo...@apache.org>
Reviewers: Xinyu Liu <xi...@apache.org>
Closes #192 from sborya/StandAloneConfig
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/1e9ae739
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/1e9ae739
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/1e9ae739
Branch: refs/heads/master
Commit: 1e9ae73958b142de5f9c5eb22e7b61f18e0bd6f5
Parents: f41fd25
Author: Boris Shkolnik <bo...@apache.org>
Authored: Mon May 22 09:36:12 2017 -0700
Committer: Xinyu Liu <xi...@xiliu-ld.linkedin.biz>
Committed: Mon May 22 09:36:12 2017 -0700
----------------------------------------------------------------------
.../versioned/jobs/configuration-table.html | 48 ++++++++++++++++++++
.../java/org/apache/samza/config/ZkConfig.java | 14 +++---
.../zk/TestZkBarrierForVersionUpgrade.java | 2 +-
3 files changed, 56 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/1e9ae739/docs/learn/documentation/versioned/jobs/configuration-table.html
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/jobs/configuration-table.html b/docs/learn/documentation/versioned/jobs/configuration-table.html
index afa42f5..fe1580f 100644
--- a/docs/learn/documentation/versioned/jobs/configuration-table.html
+++ b/docs/learn/documentation/versioned/jobs/configuration-table.html
@@ -409,6 +409,53 @@
<a href="#stores-changelog" class="property">stores.store-name.changelog</a>.
</td>
</tr>
+ <tr>
+ <td class="property" id="job.coordinator.factory">job.coordinator.factory</td>
+ <td class="default"></td>
+ <td class="description">
+ Class to use for job coordination. Currently available values are:
+ <dl>
+ <dt><code>org.apache.samza.standalone.StandaloneJobCoordinatorFactory</code></dt>
+ <dd>Fixed partition mapping. No Zoookeeper. </dd>
+ <dt><code>org.apache.samza.zk.ZkJobCoordinatorFactory</code></dt>
+ <dd>Zookeeper-based coordination. </dd>
+ </dl>
+ Required only for non-cluster-managed applications. Please see the required value for <a href=#task-name-grouper-factory>task-name-grouper-factory </a>
+ </td>
+ </tr>
+
+ <tr>
+ <!-- change link to StandAlone design/tutorial doc. SAMZA-1299 -->
+ <th colspan="3" class="section" id="ZkBasedJobCoordination"><a href="../index.html">Zookeeper-based job configuration</a></th>
+ </tr>
+ <tr>
+ <td class="property" id="job.coordinator.zk.connect">job.coordinator.zk.connect</td>
+ <td class="default"></td>
+ <td class="description">
+ <strong>Required</strong> for applications with Zookeeper-based coordination. Zookeeper coordinates (in "host:port[/znode]" format) to be used for coordination.
+ </td>
+ </tr>
+ <tr>
+ <td class="property" id="job.coordinator.zk.session.timeout.ms">job.coordinator.zk.session.timeout.ms</td>
+ <td class="default"> 30000 </td>
+ <td class="description">
+ Zookeeper session timeout for all the ZK connections in milliseconds. Session timeout controls how long zk client will wait before throwing an exception, when it cannot talk to one of ZK servers.
+ </td>
+ </tr>
+ <tr>
+ <td class="property" id="job.coordinator.zk.connection.timeout.ms">job.coordinator.zk.connection.timeout.ms</td>
+ <td class="default"> 60000 </td>
+ <td class="description">
+ Zookeeper connection timeout in milliseconds. Zk connection timeout controls how long client tries to connect to ZK server before giving up.
+ </td>
+ </tr>
+ <tr>
+ <td class="property" id="job.coordinator.zk.new.consensus.timeout.ms">job.coordinator.zk.consensus.timeout.ms</td>
+ <td class="default"> 40000 </td>
+ <td class="description">
+ How long each processor will wait for all the processors to report acceptance of the new job model before rolling back.
+ </td>
+ </tr>
<tr>
<th colspan="3" class="section" id="task"><a href="../api/overview.html">Task configuration</a></th>
</tr>
@@ -647,6 +694,7 @@
The fully-qualified name of the Java class which determines the factory class which will build the TaskNameGrouper.
The default configuration value if the property is not present is <code>task.name.grouper.factory=org.apache.samza.container.grouper.task.GroupByContainerCountFactory</code>.<br>
The user can specify a custom implementation of the TaskNameGrouperFactory where a custom logic is implemented for grouping the tasks.
+ <p><strong>Note:</strong> For non-cluster applications (ones using coordination service) one must use <i>org.apache.samza.container.grouper.task.GroupByContainerIdsFactory</i>
</td>
</tr>
http://git-wip-us.apache.org/repos/asf/samza/blob/1e9ae739/samza-core/src/main/java/org/apache/samza/config/ZkConfig.java
----------------------------------------------------------------------
diff --git a/samza-core/src/main/java/org/apache/samza/config/ZkConfig.java b/samza-core/src/main/java/org/apache/samza/config/ZkConfig.java
index fc483eb..34d2542 100644
--- a/samza-core/src/main/java/org/apache/samza/config/ZkConfig.java
+++ b/samza-core/src/main/java/org/apache/samza/config/ZkConfig.java
@@ -21,15 +21,15 @@ package org.apache.samza.config;
public class ZkConfig extends MapConfig {
// Connection string for ZK, format: :<hostname>:<port>,..."
- public static final String ZK_CONNECT = "coordinator.zk.connect";
- public static final String ZK_SESSION_TIMEOUT_MS = "coordinator.zk.session-timeout-ms";
- public static final String ZK_CONNECTION_TIMEOUT_MS = "coordinator.zk.connection-timeout-ms";
+ public static final String ZK_CONNECT = "job.coordinator.zk.connect";
+ public static final String ZK_SESSION_TIMEOUT_MS = "job.coordinator.zk.session.timeout.ms";
+ public static final String ZK_CONNECTION_TIMEOUT_MS = "job.coordinator.zk.connection.timeout.ms";
+ public static final String ZK_CONSENSUS_TIMEOUT_MS = "job.coordinator.zk.consensus.timeout.ms";
public static final int DEFAULT_CONNECTION_TIMEOUT_MS = 60000;
public static final int DEFAULT_SESSION_TIMEOUT_MS = 30000;
- public static final String ZK_BARRIER_TIMEOUT_MS = "coordinator.zk.barrier-timeout-ms";
- public static final int DEFAULT_BARRIER_TIMEOUT_MS = 40000;
-
+ public static final int DEFAULT_CONSENSUS_TIMEOUT_MS = 40000;
+
public ZkConfig(Config config) {
super(config);
}
@@ -50,6 +50,6 @@ public class ZkConfig extends MapConfig {
}
public int getZkBarrierTimeoutMs() {
- return getInt(ZK_BARRIER_TIMEOUT_MS, DEFAULT_BARRIER_TIMEOUT_MS);
+ return getInt(ZK_CONSENSUS_TIMEOUT_MS, DEFAULT_CONSENSUS_TIMEOUT_MS);
}
}
http://git-wip-us.apache.org/repos/asf/samza/blob/1e9ae739/samza-core/src/test/java/org/apache/samza/zk/TestZkBarrierForVersionUpgrade.java
----------------------------------------------------------------------
diff --git a/samza-core/src/test/java/org/apache/samza/zk/TestZkBarrierForVersionUpgrade.java b/samza-core/src/test/java/org/apache/samza/zk/TestZkBarrierForVersionUpgrade.java
index 547e32b..9c91fd3 100644
--- a/samza-core/src/test/java/org/apache/samza/zk/TestZkBarrierForVersionUpgrade.java
+++ b/samza-core/src/test/java/org/apache/samza/zk/TestZkBarrierForVersionUpgrade.java
@@ -55,7 +55,7 @@ public class TestZkBarrierForVersionUpgrade {
String processorId = "p1";
Map<String, String> map = new HashMap<>();
map.put(ZkConfig.ZK_CONNECT, testZkConnectionString);
- map.put(ZkConfig.ZK_BARRIER_TIMEOUT_MS, "200");
+ map.put(ZkConfig.ZK_CONSENSUS_TIMEOUT_MS, "200");
Config config = new MapConfig(map);
CoordinationServiceFactory serviceFactory = new ZkCoordinationServiceFactory();