You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by ja...@apache.org on 2016/09/29 18:57:20 UTC
[1/2] samza git commit: SAMZA-1020: Remove methods in Kafka
checkpoint classes that were deprecated in the 0.10 release.
Repository: samza
Updated Branches:
refs/heads/master 985822028 -> 317b6ff1b
SAMZA-1020: Remove methods in Kafka checkpoint classes that were deprecated in the 0.10 release.
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/db31e899
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/db31e899
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/db31e899
Branch: refs/heads/master
Commit: db31e8992c7bcbe17a5ad85bbab6105e232a626a
Parents: 9858220
Author: vjagadish1989 <jv...@linkedin.com>
Authored: Thu Sep 29 11:48:22 2016 -0700
Committer: vjagadish1989 <jv...@linkedin.com>
Committed: Thu Sep 29 11:48:22 2016 -0700
----------------------------------------------------------------------
.../kafka/KafkaCheckpointLogKey.scala | 23 --------------------
.../kafka/KafkaCheckpointManager.scala | 22 -------------------
.../kafka/TeskKafkaCheckpointLogKey.scala | 10 ---------
3 files changed, 55 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/db31e899/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointLogKey.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointLogKey.scala b/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointLogKey.scala
index ea8462d..9ed64c3 100644
--- a/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointLogKey.scala
+++ b/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointLogKey.scala
@@ -57,13 +57,6 @@ class KafkaCheckpointLogKey private (val map: Map[String, String]) {
*/
def isCheckpointKey = getKey.equals(CHECKPOINT_KEY_TYPE)
- /**
- * Is this key for a changelog partition mapping?
- *
- * @return true iff this key's entry is for a changelog partition mapping
- */
- @Deprecated
- def isChangelogPartitionMapping = getKey.equals(CHANGELOG_PARTITION_KEY_TYPE)
/**
* If this Key is for a checkpoint entry, return its associated TaskName.
@@ -98,17 +91,9 @@ object KafkaCheckpointLogKey {
val CHECKPOINT_KEY_KEY = "type"
val CHECKPOINT_KEY_TYPE = "checkpoint"
- @Deprecated
- val CHANGELOG_PARTITION_KEY_TYPE = "changelog-partition-mapping"
-
val CHECKPOINT_TASKNAME_KEY = "taskName"
val SYSTEMSTREAMPARTITION_GROUPER_FACTORY_KEY = "systemstreampartition-grouper-factory"
- /**
- * Partition mapping keys have no dynamic values, so we just need one instance.
- */
- @Deprecated
- val CHANGELOG_PARTITION_MAPPING_KEY = new KafkaCheckpointLogKey(Map(CHECKPOINT_KEY_KEY -> CHANGELOG_PARTITION_KEY_TYPE))
private val JSON_MAPPER = new ObjectMapper()
val KEY_TYPEREFERENCE = new TypeReference[util.HashMap[String, String]]() {}
@@ -146,14 +131,6 @@ object KafkaCheckpointLogKey {
}
/**
- * Build a key for a changelog partition mapping entry
- *
- * @return Key for changelog partition mapping entry
- */
- @Deprecated
- def getChangelogPartitionMappingKey() = CHANGELOG_PARTITION_MAPPING_KEY
-
- /**
* Deserialize a Kafka checkpoint log key
* @param bytes Serialized (via JSON) Kafka checkpoint log key
* @return Checkpoint log key
http://git-wip-us.apache.org/repos/asf/samza/blob/db31e899/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala b/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala
index ea10cae..8f18a92 100644
--- a/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala
+++ b/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala
@@ -290,28 +290,6 @@ class KafkaCheckpointManager(
}
}
-
- /**
- * Read through entire log, discarding checkpoints, finding latest changelogPartitionMapping
- * To be used for Migration purpose only. In newer version, changelogPartitionMapping will be handled through coordinator stream
- */
- @Deprecated
- def readChangeLogPartitionMapping(): util.Map[TaskName, java.lang.Integer] = {
- var changelogPartitionMapping: util.Map[TaskName, java.lang.Integer] = new util.HashMap[TaskName, java.lang.Integer]()
-
- def shouldHandleEntry(key: KafkaCheckpointLogKey) = key.isChangelogPartitionMapping
-
- def handleCheckpoint(payload: ByteBuffer, checkpointKey:KafkaCheckpointLogKey): Unit = {
- changelogPartitionMapping = serde.changelogPartitionMappingFromBytes(Utils.readBytes(payload))
-
- debug("Adding changelog partition mapping" + changelogPartitionMapping)
- }
-
- readLog(CHANGELOG_PARTITION_MAPPING_LOG4j, shouldHandleEntry, handleCheckpoint)
-
- changelogPartitionMapping
- }
-
override def toString = "KafkaCheckpointManager [systemName=%s, checkpointTopic=%s]" format(systemName, checkpointTopic)
}
http://git-wip-us.apache.org/repos/asf/samza/blob/db31e899/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TeskKafkaCheckpointLogKey.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TeskKafkaCheckpointLogKey.scala b/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TeskKafkaCheckpointLogKey.scala
index c360b6c..38f3dc2 100644
--- a/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TeskKafkaCheckpointLogKey.scala
+++ b/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TeskKafkaCheckpointLogKey.scala
@@ -40,16 +40,6 @@ class TestKafkaCheckpointLogKey {
}
@Test
- def changelogPartitionMappingKeySerializationRoundTrip() {
- val key = KafkaCheckpointLogKey.getChangelogPartitionMappingKey()
- val asBytes = key.toBytes()
- val backFromBytes = KafkaCheckpointLogKey.fromBytes(asBytes)
-
- assertEquals(key, backFromBytes)
- assertNotSame(key, backFromBytes)
- }
-
- @Test
def differingSSPGrouperFactoriesCauseException() {
val checkpointKey = KafkaCheckpointLogKey.getCheckpointKey(new TaskName("TN"))
[2/2] samza git commit: SAMZA-995: Remove deprecated property
yarn.container.count from config table.
Posted by ja...@apache.org.
SAMZA-995: Remove deprecated property yarn.container.count from config table.
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/317b6ff1
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/317b6ff1
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/317b6ff1
Branch: refs/heads/master
Commit: 317b6ff1be8b497cd1468fddf3975b2832cf9d74
Parents: db31e89
Author: Xinyu Liu <xi...@xiliu-ld.linkedin.biz>
Authored: Thu Sep 29 11:50:42 2016 -0700
Committer: vjagadish1989 <jv...@linkedin.com>
Committed: Thu Sep 29 11:54:49 2016 -0700
----------------------------------------------------------------------
docs/contribute/coding-guide.md | 2 +-
.../documentation/versioned/container/coordinator-stream.md | 2 +-
.../documentation/versioned/jobs/configuration-table.html | 9 ---------
docs/learn/documentation/versioned/jobs/reprocessing.md | 2 +-
docs/learn/documentation/versioned/jobs/yarn-jobs.md | 2 +-
docs/learn/tutorials/versioned/remote-debugging-samza.md | 2 +-
6 files changed, 5 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/317b6ff1/docs/contribute/coding-guide.md
----------------------------------------------------------------------
diff --git a/docs/contribute/coding-guide.md b/docs/contribute/coding-guide.md
index 6cb4cee..659834f 100644
--- a/docs/contribute/coding-guide.md
+++ b/docs/contribute/coding-guide.md
@@ -91,7 +91,7 @@ We are following the style guide given here (though not perfectly). Below are so
* All configuration names that define a class must end with .class (e.g. task.command.class).
* All configuration names that define a factory class must end with .factory.class (e.g. systems.kafka.consumer.factory.class).
* Configuration will always be defined as simple key/value pairs (e.g. a=b).
-* When configuration is related, it must be grouped using the same prefix (e.g. yarn.container.count=1, yarn.container.memory.bytes=1073741824).
+* When configuration is related, it must be grouped using the same prefix (e.g. job.container.count=1, yarn.container.memory.bytes=1073741824).
* When configuration must be defined multiple times, the key should be parameterized (e.g. systems.kafka.consumer.factory=x, systems.kestrel.consumer.factory=y). *When such configuration must be referred to, its parameter should be used (e.g. foo.bar.system=kafka, foo.bar.system=kestrel).
* All getter methods must be a camel case match with their configuration names (e.g. yarn.package.uri and getYarnPackageUri).
* Reading configuration should only be done in factories and main methods. Don't pass Config objects around.
http://git-wip-us.apache.org/repos/asf/samza/blob/317b6ff1/docs/learn/documentation/versioned/container/coordinator-stream.md
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/container/coordinator-stream.md b/docs/learn/documentation/versioned/container/coordinator-stream.md
index b680593..6b14960 100644
--- a/docs/learn/documentation/versioned/container/coordinator-stream.md
+++ b/docs/learn/documentation/versioned/container/coordinator-stream.md
@@ -116,7 +116,7 @@ Samza provides a command line tool to write Job Configuration messages to the co
samza-example/target/bin/run-coordinator-stream-writer.sh \
--config-path=file:///path/to/job/config.properties \
--type set-config \
- --key yarn.container.count \
+ --key job.container.count \
--value 8
{% endhighlight %}
http://git-wip-us.apache.org/repos/asf/samza/blob/317b6ff1/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 1d16f52..f60cd50 100644
--- a/docs/learn/documentation/versioned/jobs/configuration-table.html
+++ b/docs/learn/documentation/versioned/jobs/configuration-table.html
@@ -1524,15 +1524,6 @@
</tr>
<tr>
- <td class="property" id="yarn-container-count">yarn.container.count</td>
- <td class="default">1</td>
- <td class="description">
- This is deprecated in favor of
- <a href="#job-container-count" class="property">job.container.count</a>
- </td>
- </tr>
-
- <tr>
<td class="property" id="yarn-container-memory-mb">yarn.container.memory.mb</td>
<td class="default">1024</td>
<td class="description">
http://git-wip-us.apache.org/repos/asf/samza/blob/317b6ff1/docs/learn/documentation/versioned/jobs/reprocessing.md
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/jobs/reprocessing.md b/docs/learn/documentation/versioned/jobs/reprocessing.md
index 174b46f..97d8801 100644
--- a/docs/learn/documentation/versioned/jobs/reprocessing.md
+++ b/docs/learn/documentation/versioned/jobs/reprocessing.md
@@ -44,7 +44,7 @@ Normally, when a Samza job starts up, it reads the latest checkpoint to determin
1. You can stop the job, manipulate its last checkpoint to point to an older offset, and start the job up again. Samza includes a command-line tool called [CheckpointTool](../container/checkpointing.html#toc_0) which you can use to manipulate checkpoints.
2. You can start a new job with a different *job.name* or *job.id* (e.g. increment *job.id* every time you need to jump back in time). This gives the job a new checkpoint stream, with none of the old checkpoint information. You also need to set [samza.offset.default=oldest](../container/checkpointing.html), so that when the job starts up without checkpoint, it starts consuming at the oldest offset available.
-With either of these approaches you can get Samza to reprocess the entire history of messages in the input system. Input systems such as Kafka can retain a large amount of history — see discussion below. In order to speed up the reprocessing of historical data, you can increase the container count (*yarn.container.count* if you're running Samza on YARN) to boost your job's computational resources.
+With either of these approaches you can get Samza to reprocess the entire history of messages in the input system. Input systems such as Kafka can retain a large amount of history — see discussion below. In order to speed up the reprocessing of historical data, you can increase the container count (*job.container.count* if you're running Samza on YARN) to boost your job's computational resources.
If your job maintains any [persistent state](../container/state-management.html), you need to be careful when jumping back in time: resetting a checkpoint does not automatically change persistent state, so you could end up reprocessing old messages while using state from a later point in time. In most cases, a job that jumps back in time should start with an empty state. You can reset the state by deleting the changelog topic, or by changing the name of the changelog topic in your job configuration.
http://git-wip-us.apache.org/repos/asf/samza/blob/317b6ff1/docs/learn/documentation/versioned/jobs/yarn-jobs.md
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/jobs/yarn-jobs.md b/docs/learn/documentation/versioned/jobs/yarn-jobs.md
index d0e5991..821173f 100644
--- a/docs/learn/documentation/versioned/jobs/yarn-jobs.md
+++ b/docs/learn/documentation/versioned/jobs/yarn-jobs.md
@@ -37,6 +37,6 @@ yarn.queue=root.adhoc
Samza job will run in the root.adhoc queue.
-<!-- TODO document yarn.container.count and other key configs -->
+<!-- TODO document job.container.count and other key configs -->
## [Logging »](logging.html)
http://git-wip-us.apache.org/repos/asf/samza/blob/317b6ff1/docs/learn/tutorials/versioned/remote-debugging-samza.md
----------------------------------------------------------------------
diff --git a/docs/learn/tutorials/versioned/remote-debugging-samza.md b/docs/learn/tutorials/versioned/remote-debugging-samza.md
index ad1d04d..40db31a 100644
--- a/docs/learn/tutorials/versioned/remote-debugging-samza.md
+++ b/docs/learn/tutorials/versioned/remote-debugging-samza.md
@@ -62,7 +62,7 @@ task.opts=-agentlib:jdwp=transport=dt_socket,address=localhost:9009,server=y,sus
The [task.opts](../../documentation/{{site.version}}/jobs/configuration-table.html) configuration parameter is a way to override Java parameters at runtime for your Samza containers. In this example, we're setting the agentlib parameter to enable remote debugging on localhost, port 9009. In a more realistic environment, you might also set Java heap settings (-Xmx, -Xms, etc), as well as garbage collection and logging settings.
-*NOTE: If you're running multiple Samza containers on the same machine, there is a potential for port collisions. You must configure your task.opts to assign different ports for different Samza jobs. If a Samza job has more than one container (e.g. if you're using YARN with yarn.container.count=2), those containers must be run on different machines.*
+*NOTE: If you're running multiple Samza containers on the same machine, there is a potential for port collisions. You must configure your task.opts to assign different ports for different Samza jobs. If a Samza job has more than one container (e.g. if you're using YARN with job.container.count=2), those containers must be run on different machines.*
### Start the Grid