You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by cr...@apache.org on 2015/04/08 01:25:33 UTC
[01/36] samza git commit: DOCS; updating wiki links to point to cwiki
Repository: samza
Updated Branches:
refs/heads/samza-sql 7267ad37d -> 1b66a91b4
DOCS; updating wiki links to point to cwiki
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/57075c40
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/57075c40
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/57075c40
Branch: refs/heads/samza-sql
Commit: 57075c402379a9ba49a44371bc88f34a295d600b
Parents: c6fa8ed
Author: Chris Riccomini <cr...@apache.org>
Authored: Mon Mar 2 11:16:05 2015 -0800
Committer: Chris Riccomini <cr...@apache.org>
Committed: Mon Mar 2 11:16:05 2015 -0800
----------------------------------------------------------------------
docs/_layouts/default.html | 10 +++++-----
1 file changed, 5 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/57075c40/docs/_layouts/default.html
----------------------------------------------------------------------
diff --git a/docs/_layouts/default.html b/docs/_layouts/default.html
index 6f39654..13172e7 100644
--- a/docs/_layouts/default.html
+++ b/docs/_layouts/default.html
@@ -73,9 +73,9 @@
<li><a href="/learn/documentation/{{ navLink }}/jobs/configuration-table.html">Configuration</a></li>
<li><a href="/learn/documentation/{{ navLink }}/api/javadocs/">Javadocs</a></li>
<li><a href="/learn/tutorials/{{ navLink }}">Tutorials</a></li>
- <li><a href="http://wiki.apache.org/samza/FAQ">FAQ</a></li>
- <li><a href="http://wiki.apache.org/samza">Wiki</a></li>
- <li><a href="http://wiki.apache.org/samza/PapersAndTalks">Papers & Talks</a></li>
+ <li><a href="https://cwiki.apache.org/confluence/display/SAMZA/FAQ">FAQ</a></li>
+ <li><a href="https://cwiki.apache.org/confluence/display/SAMZA/Apache+Samza">Wiki</a></li>
+ <li><a href="https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=51812876">Papers & Talks</a></li>
<li><a href="http://blogs.apache.org/samza">Blog</a></li>
</ul>
@@ -84,8 +84,8 @@
<li><a href="/community/mailing-lists.html">Mailing Lists</a></li>
<li><a href="/community/irc.html">IRC</a></li>
<li><a href="https://issues.apache.org/jira/browse/SAMZA">Bugs</a></li>
- <li><a href="http://wiki.apache.org/samza/PoweredBy">Powered by</a></li>
- <li><a href="http://wiki.apache.org/samza/Ecosystem">Ecosystem</a></li>
+ <li><a href="https://cwiki.apache.org/confluence/display/SAMZA/Powered+By">Powered by</a></li>
+ <li><a href="https://cwiki.apache.org/confluence/display/SAMZA/Ecosystem">Ecosystem</a></li>
<li><a href="/community/committers.html">Committers</a></li>
</ul>
[32/36] samza git commit: SAMZA-633;
expose configuration to Kafka's encoder and decoder
Posted by cr...@apache.org.
SAMZA-633; expose configuration to Kafka's encoder and decoder
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/76e151c1
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/76e151c1
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/76e151c1
Branch: refs/heads/samza-sql
Commit: 76e151c14a5d01b88c49edfb2c847ffe3c0e6242
Parents: cde495e
Author: Dan Harvey <da...@gmail.com>
Authored: Thu Apr 2 16:07:56 2015 -0700
Committer: Chris Riccomini <cr...@apache.org>
Committed: Thu Apr 2 16:07:56 2015 -0700
----------------------------------------------------------------------
.../apache/samza/config/KafkaSerdeConfig.scala | 12 ++++++++++++
.../apache/samza/serializers/KafkaSerde.scala | 20 ++++++++++++++++++--
.../samza/config/TestKafkaSerdeConfig.scala | 7 +++++++
3 files changed, 37 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/76e151c1/samza-kafka/src/main/scala/org/apache/samza/config/KafkaSerdeConfig.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/scala/org/apache/samza/config/KafkaSerdeConfig.scala b/samza-kafka/src/main/scala/org/apache/samza/config/KafkaSerdeConfig.scala
index 11078e3..ed2bd40 100644
--- a/samza-kafka/src/main/scala/org/apache/samza/config/KafkaSerdeConfig.scala
+++ b/samza-kafka/src/main/scala/org/apache/samza/config/KafkaSerdeConfig.scala
@@ -19,6 +19,10 @@
package org.apache.samza.config
+import java.util.Properties
+
+import kafka.utils.VerifiableProperties
+
object KafkaSerdeConfig {
// kafka serde config constants
val ENCODER = SerializerConfig.SERIALIZER_PREFIX + ".encoder"
@@ -33,4 +37,12 @@ class KafkaSerdeConfig(config: Config) extends ScalaMapConfig(config) {
def getKafkaDecoder(serializer: String) =
getOption(KafkaSerdeConfig.DECODER format serializer)
+
+ def getKafkaProperties(serializer: String) = {
+ val properties = new Properties();
+ val prefix = SerializerConfig.SERIALIZER_PREFIX format serializer + "."
+ properties.putAll(config.subset(prefix, true))
+
+ new VerifiableProperties(properties);
+ }
}
http://git-wip-us.apache.org/repos/asf/samza/blob/76e151c1/samza-kafka/src/main/scala/org/apache/samza/serializers/KafkaSerde.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/scala/org/apache/samza/serializers/KafkaSerde.scala b/samza-kafka/src/main/scala/org/apache/samza/serializers/KafkaSerde.scala
index 82ba2a0..d626b79 100644
--- a/samza-kafka/src/main/scala/org/apache/samza/serializers/KafkaSerde.scala
+++ b/samza-kafka/src/main/scala/org/apache/samza/serializers/KafkaSerde.scala
@@ -22,6 +22,7 @@ import java.nio.ByteBuffer
import org.apache.samza.util.Util
import kafka.serializer.Encoder
import kafka.serializer.Decoder
+import kafka.utils.VerifiableProperties
import org.apache.samza.config.Config
import org.apache.samza.config.KafkaSerdeConfig.Config2KafkaSerde
import org.apache.samza.SamzaException
@@ -39,8 +40,23 @@ class KafkaSerdeFactory[T] extends SerdeFactory[T] {
val decoderClassName = config
.getKafkaDecoder(name)
.getOrElse(throw new SamzaException("No kafka decoder defined for %s" format name))
- val encoder = Util.getObj[Encoder[T]](encoderClassName)
- val decoder = Util.getObj[Decoder[T]](decoderClassName)
+
+ val verifiableProperties = config.getKafkaProperties(name)
+
+ val encoder = getObj[Encoder[T]](encoderClassName, verifiableProperties);
+ val decoder = getObj[Decoder[T]](decoderClassName, verifiableProperties);
+
new KafkaSerde(encoder, decoder)
}
+
+ /**
+ * Instantiate a class instance from a given className and properties.
+ */
+ private def getObj[T](className: String, properties: VerifiableProperties) = {
+ Class
+ .forName(className)
+ .getDeclaredConstructor(classOf[VerifiableProperties])
+ .newInstance(properties)
+ .asInstanceOf[T]
+ }
}
http://git-wip-us.apache.org/repos/asf/samza/blob/76e151c1/samza-kafka/src/test/scala/org/apache/samza/config/TestKafkaSerdeConfig.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/test/scala/org/apache/samza/config/TestKafkaSerdeConfig.scala b/samza-kafka/src/test/scala/org/apache/samza/config/TestKafkaSerdeConfig.scala
index 5cf82c2..d6899b8 100644
--- a/samza-kafka/src/test/scala/org/apache/samza/config/TestKafkaSerdeConfig.scala
+++ b/samza-kafka/src/test/scala/org/apache/samza/config/TestKafkaSerdeConfig.scala
@@ -38,4 +38,11 @@ class TestKafkaSerdeConfig {
assertEquals(MAGIC_VAL, config.getKafkaEncoder("test").getOrElse(""))
assertEquals(MAGIC_VAL, config.getKafkaDecoder("test").getOrElse(""))
}
+
+ @Test
+ def testKafkaConfigurationProperties {
+ val properties = config.getKafkaProperties("test")
+ assertEquals(MAGIC_VAL, properties.getString("encoder"))
+ assertEquals(MAGIC_VAL, properties.getString("decoder"))
+ }
}
[36/36] samza git commit: Merge branch 'master' into samza-sql
Posted by cr...@apache.org.
Merge branch 'master' into samza-sql
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/1b66a91b
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/1b66a91b
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/1b66a91b
Branch: refs/heads/samza-sql
Commit: 1b66a91b4772266394b93ed807e9c940151243f9
Parents: 7267ad3 edc2c78
Author: Chris Riccomini <cr...@apache.org>
Authored: Tue Apr 7 16:20:01 2015 -0700
Committer: Chris Riccomini <cr...@apache.org>
Committed: Tue Apr 7 16:20:01 2015 -0700
----------------------------------------------------------------------
.travis.yml | 46 ----
KEYS | 60 ++++-
README.md | 2 -
build.gradle | 11 +
docs/_config.yml | 2 +-
docs/_layouts/default.html | 11 +-
docs/archive/index.html | 8 +
docs/contribute/tests.md | 6 -
.../versioned/container/metrics.md | 2 +-
.../versioned/jobs/configuration-table.html | 13 +
.../versioned/jobs/configuration.md | 5 +-
.../documentation/versioned/jobs/job-runner.md | 2 +-
.../documentation/versioned/jobs/logging.md | 13 +-
.../versioned/run-in-multi-node-yarn.md | 14 +-
docs/startup/download/index.md | 25 +-
gradle.properties | 2 +-
gradle/dependency-versions.gradle | 4 +-
gradle/release.gradle | 4 +-
.../java/org/apache/samza/config/Config.java | 5 +
.../java/org/apache/samza/config/MapConfig.java | 21 +-
.../java/org/apache/samza/task/TaskContext.java | 13 +
.../org/apache/samza/config/TestConfig.java | 21 +-
.../apache/samza/config/SerializerConfig.scala | 3 +-
.../org/apache/samza/config/StorageConfig.scala | 16 +-
.../org/apache/samza/container/RunLoop.scala | 31 ++-
.../apache/samza/container/SamzaContainer.scala | 8 +-
.../apache/samza/container/TaskInstance.scala | 6 +
.../samza/serializers/ByteBufferSerde.scala | 49 ++++
.../apache/samza/serializers/ByteSerde.scala | 3 +-
.../apache/samza/serializers/StringSerde.scala | 1 -
.../apache/samza/serializers/UUIDSerde.scala | 47 ++++
.../apache/samza/system/SystemConsumers.scala | 7 +-
.../apache/samza/config/TestStorageConfig.scala | 40 +++
.../apache/samza/container/TestRunLoop.scala | 38 ++-
.../samza/container/TestSamzaContainer.scala | 5 +-
.../samza/container/TestTaskInstance.scala | 68 ++++-
.../samza/serializers/TestByteBufferSerde.scala | 41 ++++
.../samza/serializers/TestIntegerSerde.scala | 3 -
.../samza/serializers/TestUUIDSerde.scala | 53 ++++
.../samza/system/TestStreamMetadataCache.scala | 4 +-
.../kafka/KafkaCheckpointManager.scala | 7 +-
.../kafka/KafkaCheckpointManagerFactory.scala | 4 +-
.../apache/samza/config/KafkaSerdeConfig.scala | 12 +
.../apache/samza/serializers/KafkaSerde.scala | 20 +-
.../apache/samza/system/kafka/BrokerProxy.scala | 31 ++-
.../apache/samza/system/kafka/GetOffset.scala | 7 +-
.../samza/system/kafka/KafkaSystemAdmin.scala | 10 +-
.../system/kafka/KafkaSystemConsumer.scala | 27 +-
.../samza/system/kafka/KafkaSystemFactory.scala | 12 +-
.../system/kafka/KafkaSystemProducer.scala | 10 +-
.../samza/system/kafka/TopicMetadataCache.scala | 12 +-
.../scala/org/apache/samza/util/KafkaUtil.scala | 39 ++-
.../samza/system/kafka/MockKafkaProducer.java | 246 +++++++++++++++++++
.../java/org/apache/samza/utils/TestUtils.java | 112 +++++++++
.../samza/config/TestKafkaSerdeConfig.scala | 7 +
.../samza/system/kafka/MockKafkaProducer.java | 246 -------------------
.../samza/system/kafka/TestBrokerProxy.scala | 59 +++++
.../system/kafka/TestKafkaSystemAdmin.scala | 8 +-
.../system/kafka/TestKafkaSystemFactory.scala | 13 +
.../system/kafka/TestKafkaSystemProducer.scala | 33 ++-
.../system/kafka/TestTopicMetadataCache.scala | 14 +-
.../org/apache/samza/utils/TestKafkaUtil.scala | 33 +++
.../scala/org/apache/samza/utils/TestUtils.java | 112 ---------
.../RocksDbKeyValueStorageEngineFactory.scala | 4 +-
.../samza/storage/kv/RocksDbKeyValueStore.scala | 11 +-
.../apache/samza/storage/kv/CachedStore.scala | 15 ++
.../samza/storage/kv/TestCachedStore.scala | 35 +++
.../apache/samza/config/Log4jSystemConfig.java | 67 +++--
.../samza/logging/log4j/StreamAppender.java | 36 +--
.../serializers/LoggingEventJsonSerde.java | 192 +++++++++++++++
.../LoggingEventJsonSerdeFactory.java | 38 +++
.../samza/config/TestLog4jSystemConfig.java | 22 +-
.../samza/logging/log4j/TestStreamAppender.java | 54 +++-
samza-test/src/main/config/join/checker.samza | 1 +
.../src/main/config/perf/kv-perf.properties | 28 ++-
samza-test/src/main/python/configs/tests.json | 2 +-
.../performance/TestKeyValuePerformance.scala | 175 ++++++++-----
.../test/integration/TestStatefulTask.scala | 8 +-
.../webapp/ApplicationMasterRestServlet.scala | 2 +-
.../webapp/ApplicationMasterWebServlet.scala | 2 +-
80 files changed, 1790 insertions(+), 689 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/1b66a91b/build.gradle
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/1b66a91b/gradle/dependency-versions.gradle
----------------------------------------------------------------------
[14/36] samza git commit: SAMZA-599; upgrade to kafka 0.8.2.1
Posted by cr...@apache.org.
SAMZA-599; upgrade to kafka 0.8.2.1
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/81701498
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/81701498
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/81701498
Branch: refs/heads/samza-sql
Commit: 817014985b24cda4756437a76592052c39aff2c1
Parents: 7e4f3e4
Author: Chris Riccomini <cr...@apache.org>
Authored: Mon Mar 16 13:24:42 2015 -0700
Committer: Chris Riccomini <cr...@apache.org>
Committed: Mon Mar 16 13:24:42 2015 -0700
----------------------------------------------------------------------
gradle/dependency-versions.gradle | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/81701498/gradle/dependency-versions.gradle
----------------------------------------------------------------------
diff --git a/gradle/dependency-versions.gradle b/gradle/dependency-versions.gradle
index cb6ef21..ee6dfc4 100644
--- a/gradle/dependency-versions.gradle
+++ b/gradle/dependency-versions.gradle
@@ -26,7 +26,7 @@
zkClientVersion = "0.3"
zookeeperVersion = "3.3.4"
metricsVersion = "2.2.0"
- kafkaVersion = "0.8.2.0"
+ kafkaVersion = "0.8.2.1"
commonsHttpClientVersion = "3.1"
rocksdbVersion = "3.5.1"
yarnVersion = "2.4.0"
[18/36] samza git commit: SAMZA-543; disable WAL for rocksdb kv store
Posted by cr...@apache.org.
SAMZA-543; disable WAL for rocksdb kv store
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/a6ada1f6
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/a6ada1f6
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/a6ada1f6
Branch: refs/heads/samza-sql
Commit: a6ada1f6fa4fa401201ad863eda3264c1782e276
Parents: 137d0d6
Author: Navina Ramesh <na...@gmail.com>
Authored: Wed Mar 18 18:47:08 2015 -0700
Committer: Chris Riccomini <cr...@apache.org>
Committed: Wed Mar 18 18:47:08 2015 -0700
----------------------------------------------------------------------
.../RocksDbKeyValueStorageEngineFactory.scala | 4 +-
.../samza/storage/kv/RocksDbKeyValueStore.scala | 11 +-
.../src/main/config/perf/kv-perf.properties | 28 ++-
.../performance/TestKeyValuePerformance.scala | 175 ++++++++++++-------
4 files changed, 142 insertions(+), 76 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/a6ada1f6/samza-kv-rocksdb/src/main/scala/org/apache/samza/storage/kv/RocksDbKeyValueStorageEngineFactory.scala
----------------------------------------------------------------------
diff --git a/samza-kv-rocksdb/src/main/scala/org/apache/samza/storage/kv/RocksDbKeyValueStorageEngineFactory.scala b/samza-kv-rocksdb/src/main/scala/org/apache/samza/storage/kv/RocksDbKeyValueStorageEngineFactory.scala
index 14eeba5..5ab6859 100644
--- a/samza-kv-rocksdb/src/main/scala/org/apache/samza/storage/kv/RocksDbKeyValueStorageEngineFactory.scala
+++ b/samza-kv-rocksdb/src/main/scala/org/apache/samza/storage/kv/RocksDbKeyValueStorageEngineFactory.scala
@@ -24,6 +24,7 @@ import org.apache.samza.container.SamzaContainerContext
import org.apache.samza.metrics.MetricsRegistry
import org.apache.samza.storage.kv._
import org.apache.samza.system.SystemStreamPartition
+import org.rocksdb.WriteOptions
class RocksDbKeyValueStorageEngineFactory [K, V] extends BaseKeyValueStorageEngineFactory[K, V]
{
@@ -44,7 +45,8 @@ class RocksDbKeyValueStorageEngineFactory [K, V] extends BaseKeyValueStorageEngi
val storageConfig = containerContext.config.subset("stores." + storeName + ".", true)
val rocksDbMetrics = new KeyValueStoreMetrics(storeName, registry)
val rocksDbOptions = RocksDbKeyValueStore.options(storageConfig, containerContext)
- val rocksDb = new RocksDbKeyValueStore(storeDir, rocksDbOptions, rocksDbMetrics)
+ val rocksDbWriteOptions = new WriteOptions().setDisableWAL(true)
+ val rocksDb = new RocksDbKeyValueStore(storeDir, rocksDbOptions, rocksDbWriteOptions, rocksDbMetrics)
rocksDb
}
}
http://git-wip-us.apache.org/repos/asf/samza/blob/a6ada1f6/samza-kv-rocksdb/src/main/scala/org/apache/samza/storage/kv/RocksDbKeyValueStore.scala
----------------------------------------------------------------------
diff --git a/samza-kv-rocksdb/src/main/scala/org/apache/samza/storage/kv/RocksDbKeyValueStore.scala b/samza-kv-rocksdb/src/main/scala/org/apache/samza/storage/kv/RocksDbKeyValueStore.scala
index eae2a5a..66c2a0d 100644
--- a/samza-kv-rocksdb/src/main/scala/org/apache/samza/storage/kv/RocksDbKeyValueStore.scala
+++ b/samza-kv-rocksdb/src/main/scala/org/apache/samza/storage/kv/RocksDbKeyValueStore.scala
@@ -29,7 +29,7 @@ object RocksDbKeyValueStore extends Logging {
def options(storeConfig: Config, containerContext: SamzaContainerContext) = {
val cacheSize = storeConfig.getLong("container.cache.size.bytes", 100 * 1024 * 1024L)
val writeBufSize = storeConfig.getLong("container.write.buffer.size.bytes", 32 * 1024 * 1024)
- val options = new Options();
+ val options = new Options()
// Cache size and write buffer size are specified on a per-container basis.
val numTasks = containerContext.taskNames.size
@@ -77,6 +77,7 @@ object RocksDbKeyValueStore extends Logging {
class RocksDbKeyValueStore(
val dir: File,
val options: Options,
+ val writeOptions: WriteOptions = new WriteOptions(),
val metrics: KeyValueStoreMetrics = new KeyValueStoreMetrics) extends KeyValueStore[Array[Byte], Array[Byte]] with Logging {
private lazy val db = RocksDB.open(options, dir.toString)
@@ -97,11 +98,11 @@ class RocksDbKeyValueStore(
metrics.puts.inc
require(key != null, "Null key not allowed.")
if (value == null) {
- db.remove(key)
+ db.remove(writeOptions, key)
deletesSinceLastCompaction += 1
} else {
metrics.bytesWritten.inc(key.size + value.size)
- db.put(key, value)
+ db.put(writeOptions, key, value)
}
}
@@ -115,12 +116,12 @@ class RocksDbKeyValueStore(
val curr = iter.next()
if (curr.getValue == null) {
deletes += 1
- db.remove(curr.getKey);
+ db.remove(writeOptions, curr.getKey)
} else {
val key = curr.getKey
val value = curr.getValue
metrics.bytesWritten.inc(key.size + value.size)
- db.put(key, value)
+ db.put(writeOptions, key, value)
}
}
metrics.puts.inc(wrote)
http://git-wip-us.apache.org/repos/asf/samza/blob/a6ada1f6/samza-test/src/main/config/perf/kv-perf.properties
----------------------------------------------------------------------
diff --git a/samza-test/src/main/config/perf/kv-perf.properties b/samza-test/src/main/config/perf/kv-perf.properties
index 0d487b1..33fcd8d 100644
--- a/samza-test/src/main/config/perf/kv-perf.properties
+++ b/samza-test/src/main/config/perf/kv-perf.properties
@@ -15,7 +15,27 @@
# specific language governing permissions and limitations
# under the License.
-stores.test.factory=org.apache.samza.storage.kv.RocksDbKeyValueStorageEngineFactory
-stores.test.compaction.delete.threshold=1000
-test.partition.count=4
-test.num.loops=1000
+# Config for all-with-deletes
+test.all-with-deletes.stores.test-store.factory=org.apache.samza.storage.kv.RocksDbKeyValueStorageEngineFactory
+
+test.all-with-deletes.partition.count=4
+
+test.all-with-deletes.set.count=1
+test.all-with-deletes.set-1.num.loops=1000
+
+# Config for rocksdb-write-performance
+test.rocksdb-write-performance.stores.test-store.factory=org.apache.samza.storage.kv.RocksDbKeyValueStorageEngineFactory
+test.rocksdb-write-performance.partition.count=4
+
+test.rocksdb-write-performance.set.count=3
+test.rocksdb-write-performance.set-1.message.size=256
+test.rocksdb-write-performance.set-1.message.count=1000000
+
+test.rocksdb-write-performance.set-2.message.size=512
+test.rocksdb-write-performance.set-2.message.count=1000000
+
+test.rocksdb-write-performance.set-3.message.size=1024
+test.rocksdb-write-performance.set-3.message.count=1000000
+
+# List of tests to execute
+test.methods=rocksdb-write-performance
http://git-wip-us.apache.org/repos/asf/samza/blob/a6ada1f6/samza-test/src/main/scala/org/apache/samza/test/performance/TestKeyValuePerformance.scala
----------------------------------------------------------------------
diff --git a/samza-test/src/main/scala/org/apache/samza/test/performance/TestKeyValuePerformance.scala b/samza-test/src/main/scala/org/apache/samza/test/performance/TestKeyValuePerformance.scala
index b4fa7d3..0858b98 100644
--- a/samza-test/src/main/scala/org/apache/samza/test/performance/TestKeyValuePerformance.scala
+++ b/samza-test/src/main/scala/org/apache/samza/test/performance/TestKeyValuePerformance.scala
@@ -22,114 +22,134 @@ package org.apache.samza.test.performance
import org.apache.samza.util.Logging
import org.apache.samza.config.Config
import org.apache.samza.config.StorageConfig._
-import org.apache.samza.container.{ TaskName, SamzaContainerContext }
+import org.apache.samza.container.{TaskName, SamzaContainerContext}
import org.apache.samza.metrics.MetricsRegistryMap
import org.apache.samza.storage.kv.KeyValueStore
import org.apache.samza.storage.kv.KeyValueStorageEngine
import org.apache.samza.storage.StorageEngineFactory
import org.apache.samza.util.CommandLine
import org.apache.samza.util.Util
-import org.apache.samza.serializers.ByteSerde
+import org.apache.samza.serializers.{StringSerde, ByteSerde, SerdeManager}
import org.apache.samza.Partition
import org.apache.samza.SamzaException
import org.apache.samza.task.TaskInstanceCollector
import org.apache.samza.system.SystemProducers
import org.apache.samza.system.SystemProducer
-import org.apache.samza.serializers.SerdeManager
import java.io.File
import java.util.UUID
+import java.util
/**
* A simple CLI-based tool for running various key-value performance tests.
+ *
+ * List of KeyValuePerformance tests must be defined in 'test.methods' configuration as a comma-separated value.
+ * The tool splits this list to determine which tests to run.
+ *
+ * Each test should define its own set of configuration for partition count, stores etc.
+ * using the "test.<test-name>.<config-string>=<config-value>" pattern
+ *
+ * Each test may define one or more test parameterss.
+ * For example, test1 can define 2 sets of parameters by specifying "test.test1.set.count=2" and
+ * define each set as:
+ * "test.test1.set-1.<param-name>=<param-value>"
+ * "test.test1.set-2.<param-name>=<param-value>"
*/
+
object TestKeyValuePerformance extends Logging {
val Encoding = "UTF-8"
- /**
- * KeyValuePerformance job configs must define a 'test.method' configuration.
- * This configuration must be the value of one of the keys in this map. The
- * test uses this key to determine which test to run.
- */
- val testMethods: Map[String, (Config) => Unit] = Map("testAllWithDeletes" -> runTestAllWithDeletes)
+ val testMethods: Map[String, (KeyValueStorageEngine[Array[Byte], Array[Byte]], Config) => Unit] = Map(
+ "all-with-deletes" -> runTestAllWithDeletes,
+ "rocksdb-write-performance" -> runTestMsgWritePerformance
+ )
def main(args: Array[String]) {
val cmdline = new CommandLine
val options = cmdline.parser.parse(args: _*)
val config = cmdline.loadConfig(options)
- val testMethod = config.get("test.method", "testAllWithDeletes")
-
- info("Got arguments: %s" format args.toList)
- info("Using config: %s" format config)
- info("Using test method: %s" format testMethod)
-
- if (testMethods.contains(testMethod)) {
- testMethods(testMethod)(config)
- } else {
- error("Invalid test method. Valid methods are: %s" format testMethods.keys)
+ val tests = config.get("test.methods", "rocksdb-write-performance,all-with-deletes").split(",")
- throw new SamzaException("Unknown test method: %s" format testMethod)
+ tests.foreach{ test =>
+ info("Running test: %s" format test)
+ if(testMethods.contains(test)) {
+ invokeTest(test, testMethods(test), config.subset("test." + test + ".", true))
+ } else {
+ error("Invalid test method. valid methods are: %s" format testMethods.keys)
+ throw new SamzaException("Unknown test method: %s" format test)
+ }
}
}
- /**
- * Do wiring for testAllWithDeletes, and run the test.
- */
- def runTestAllWithDeletes(config: Config) {
- val test = new TestKeyValuePerformance
- val serde = new ByteSerde
- val partitionCount = config.getInt("test.partition.count", 1)
- val numLoops = config.getInt("test.num.loops", 100)
- val messagesPerBatch = config.getInt("test.messages.per.batch", 10000)
- val messageSizeBytes = config.getInt("test.message.size.bytes", 200)
- val taskNames = new java.util.ArrayList[TaskName]()
-
+ def invokeTest(testName: String, testMethod: (KeyValueStorageEngine[Array[Byte], Array[Byte]], Config) => Unit, config: Config) {
+ val taskNames = new util.ArrayList[TaskName]()
+ val partitionCount = config.getInt("partition.count", 1)
(0 until partitionCount).map(p => taskNames.add(new TaskName(new Partition(p).toString)))
- info("Using partition count: %s" format partitionCount)
- info("Using num loops: %s" format numLoops)
- info("Using messages per batch: %s" format messagesPerBatch)
- info("Using message size: %s bytes" format messageSizeBytes)
-
+ val producerMultiplexer = new SystemProducers(
+ Map[String, SystemProducer](),
+ new SerdeManager
+ )
// Build a Map[String, StorageEngineFactory]. The key is the store name.
- val storageEngineFactories = config
+ val storageEngineMappings = config
.getStoreNames
.map(storeName => {
- val storageFactoryClassName = config
- .getStorageFactoryClassName(storeName)
- .getOrElse(throw new SamzaException("Missing storage factory for %s." format storeName))
+ val storageFactoryClassName =
+ config.getStorageFactoryClassName(storeName)
+ .getOrElse(throw new SamzaException("Missing storage factory for %s." format storeName))
(storeName, Util.getObj[StorageEngineFactory[Array[Byte], Array[Byte]]](storageFactoryClassName))
- }).toMap
+ })
- val producerMultiplexer = new SystemProducers(
- Map[String, SystemProducer](),
- new SerdeManager)
+ for((storeName, storageEngine) <- storageEngineMappings) {
+ val testSetCount = config.getInt("set.count", 1)
+ (1 to testSetCount).foreach(testSet => {
+ //Create a new DB instance for each test set
+ val output = new File("/tmp/" + UUID.randomUUID())
+ val byteSerde = new ByteSerde
+ info("Using output directory %s for %s using %s." format (output, storeName, storageEngine.getClass.getCanonicalName))
+ val engine = storageEngine.getStorageEngine(
+ storeName,
+ output,
+ byteSerde,
+ byteSerde,
+ new TaskInstanceCollector(producerMultiplexer),
+ new MetricsRegistryMap,
+ null,
+ new SamzaContainerContext(0, config, taskNames)
+ )
+
+ val db = if(!engine.isInstanceOf[KeyValueStorageEngine[_,_]]) {
+ throw new SamzaException("This test can only run with KeyValueStorageEngine configured as store factory.")
+ } else {
+ engine.asInstanceOf[KeyValueStorageEngine[Array[Byte], Array[Byte]]]
+ }
+
+ // Run the test method
+ testMethod(db, config.subset("set-" + testSet + ".", true))
+
+ info("Cleaning up output directory for %s." format storeName)
+ Util.rm(output)
+ })
+ }
+ }
- for ((storeName, storageEngine) <- storageEngineFactories) {
- val output = new File("/tmp/" + UUID.randomUUID)
+ def runTestAllWithDeletes(db: KeyValueStore[Array[Byte], Array[Byte]], config: Config) {
+ val numLoops = config.getInt("num.loops", 100)
+ val messagesPerBatch = config.getInt("messages.per.batch", 10000)
+ val messageSizeBytes = config.getInt("message.size.bytes", 200)
- info("Using output directory %s for %s." format (output, storeName))
+ info("Using (num loops, messages per batch, message size in bytes) => (%s, %s, %s)" format (numLoops, messagesPerBatch, messageSizeBytes))
+ new TestKeyValuePerformance().testAllWithDeletes(db, numLoops, messagesPerBatch, messageSizeBytes)
- val engine = storageEngine.getStorageEngine(
- storeName,
- output,
- serde,
- serde,
- new TaskInstanceCollector(producerMultiplexer),
- new MetricsRegistryMap,
- null,
- new SamzaContainerContext(0, config, taskNames))
+ }
- val db = if (!engine.isInstanceOf[KeyValueStorageEngine[_, _]]) {
- throw new SamzaException("This test can only run with KeyValueStorageEngine configured as store factory.")
- } else {
- engine.asInstanceOf[KeyValueStorageEngine[Array[Byte], Array[Byte]]]
- }
+ def runTestMsgWritePerformance(db: KeyValueStore[Array[Byte], Array[Byte]], config: Config) {
+ val messageSizeBytes = config.getInt("message.size", 200)
+ val messageCount = config.getInt("message.count", 10000)
- test.testAllWithDeletes(db, numLoops, messagesPerBatch, messageSizeBytes)
- info("Cleaning up output directory for %s." format storeName)
- Util.rm(output)
- }
+ info("Using (message count, message size in bytes) => (%s, %s)" format (messageCount, messageSizeBytes))
+ new TestKeyValuePerformance().testMsgWritePerformance(db, messageCount, messageSizeBytes)
}
+
}
class TestKeyValuePerformance extends Logging {
@@ -179,4 +199,27 @@ class TestKeyValuePerformance extends Logging {
info("Total time: %ss" format ((System.currentTimeMillis - start) * .001))
}
+
+
+ /**
+ * Test that successively writes a set of fixed-size messages to the KV store
+ * and computes the total time for the operations
+ * @param store Key-Value store instance that is being tested
+ * @param numMsgs Total number of messages to write to the store
+ * @param msgSizeInBytes Size of each message in Bytes
+ */
+ def testMsgWritePerformance(
+ store: KeyValueStore[Array[Byte], Array[Byte]],
+ numMsgs: Int = 10000,
+ msgSizeInBytes: Int = 200) {
+
+ val msg = (0 until msgSizeInBytes).map(i => "x").mkString.getBytes(Encoding)
+
+ val start = System.currentTimeMillis
+ (0 until numMsgs).foreach(i => {
+ store.put(i.toString.getBytes(Encoding), msg)
+ })
+ val timeTaken = System.currentTimeMillis - start
+ info("Total time to write %d msgs of size %d bytes : %s s" format (numMsgs, msgSizeInBytes, timeTaken * .001))
+ }
}
\ No newline at end of file
[30/36] samza git commit: update download page for 0.9.0
Posted by cr...@apache.org.
update download page for 0.9.0
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/95b6afc3
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/95b6afc3
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/95b6afc3
Branch: refs/heads/samza-sql
Commit: 95b6afc305270478bfd830527175b20fc18716ec
Parents: faf93ff
Author: Yan Fang <ya...@gmail.com>
Authored: Wed Apr 1 11:31:05 2015 -0700
Committer: Yan Fang <ya...@gmail.com>
Committed: Wed Apr 1 11:31:05 2015 -0700
----------------------------------------------------------------------
docs/startup/download/index.md | 21 +++++++++++----------
1 file changed, 11 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/95b6afc3/docs/startup/download/index.md
----------------------------------------------------------------------
diff --git a/docs/startup/download/index.md b/docs/startup/download/index.md
index 47ef0e0..37f948b 100644
--- a/docs/startup/download/index.md
+++ b/docs/startup/download/index.md
@@ -29,7 +29,8 @@ Starting in June of 2015, Samza will begin requiring JDK7 or higher. Please see
### Source Releases
- * [samza-sources-0.8.0-incubating.tgz](http://www.apache.org/dyn/closer.cgi/samza/0.8.0-incubating)
+ * [samza-sources-0.9.0.tgz](http://www.apache.org/dyn/closer.cgi/samza/0.9.0)
+ * [samza-sources-0.8.0-incubating.tgz](https://archive.apache.org/dist/incubator/samza/0.8.0-incubating)
* [samza-sources-0.7.0-incubating.tgz](https://archive.apache.org/dist/incubator/samza/0.7.0-incubating)
### Maven
@@ -44,55 +45,55 @@ A Maven-based Samza project can pull in all required dependencies Samza dependen
<dependency>
<groupId>org.apache.samza</groupId>
<artifactId>samza-api</artifactId>
- <version>0.8.0</version>
+ <version>0.9.0</version>
</dependency>
<dependency>
<groupId>org.apache.samza</groupId>
<artifactId>samza-core_2.10</artifactId>
- <version>0.8.0</version>
+ <version>0.9.0</version>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.apache.samza</groupId>
<artifactId>samza-serializers_2.10</artifactId>
- <version>0.8.0</version>
+ <version>0.9.0</version>
</dependency>
<dependency>
<groupId>org.apache.samza</groupId>
<artifactId>samza-shell</artifactId>
<classifier>dist</classifier>
<type>tgz</type>
- <version>0.8.0</version>
+ <version>0.9.0</version>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.apache.samza</groupId>
<artifactId>samza-yarn_2.10</artifactId>
- <version>0.8.0</version>
+ <version>0.9.0</version>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.apache.samza</groupId>
<artifactId>samza-kv_2.10</artifactId>
- <version>0.8.0</version>
+ <version>0.9.0</version>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.apache.samza</groupId>
<artifactId>samza-kv-rocksdb_2.10</artifactId>
- <version>0.8.0</version>
+ <version>0.9.0</version>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.apache.samza</groupId>
<artifactId>samza-kv-inmemory_2.10</artifactId>
- <version>0.8.0</version>
+ <version>0.9.0</version>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.apache.samza</groupId>
<artifactId>samza-kafka_2.10</artifactId>
- <version>0.8.0</version>
+ <version>0.9.0</version>
<scope>runtime</scope>
</dependency>
{% endhighlight %}
[15/36] samza git commit: SAMZA-589;
sanitize sensitive data beore displaying configs to UI
Posted by cr...@apache.org.
SAMZA-589; sanitize sensitive data beore displaying configs to UI
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/cad2c828
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/cad2c828
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/cad2c828
Branch: refs/heads/samza-sql
Commit: cad2c828692244d110e9e3a19c39b257c3040b4c
Parents: 8170149
Author: Tommy Becker <tw...@gmail.com>
Authored: Mon Mar 16 14:53:32 2015 -0700
Committer: Chris Riccomini <cr...@apache.org>
Committed: Mon Mar 16 14:53:32 2015 -0700
----------------------------------------------------------------------
build.gradle | 1 +
.../versioned/jobs/configuration.md | 5 ++++-
.../java/org/apache/samza/config/Config.java | 5 +++++
.../java/org/apache/samza/config/MapConfig.java | 21 +++++++++++++++++---
.../org/apache/samza/config/TestConfig.java | 21 +++++++++++++++++++-
.../webapp/ApplicationMasterRestServlet.scala | 2 +-
.../webapp/ApplicationMasterWebServlet.scala | 2 +-
7 files changed, 50 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/cad2c828/build.gradle
----------------------------------------------------------------------
diff --git a/build.gradle b/build.gradle
index 113da2a..97de3a2 100644
--- a/build.gradle
+++ b/build.gradle
@@ -56,6 +56,7 @@ rat {
'**/.git/**',
'**/.gitignore',
'**/.gradle/**',
+ '**/.idea/**',
'**/.project',
'**/.settings/**',
'**/.DS_Store/**',
http://git-wip-us.apache.org/repos/asf/samza/blob/cad2c828/docs/learn/documentation/versioned/jobs/configuration.md
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/jobs/configuration.md b/docs/learn/documentation/versioned/jobs/configuration.md
index e094f60..c0ebb69 100644
--- a/docs/learn/documentation/versioned/jobs/configuration.md
+++ b/docs/learn/documentation/versioned/jobs/configuration.md
@@ -58,6 +58,9 @@ Configuration keys that absolutely must be defined for a Samza job are:
### Configuration Keys
-A complete list of configuration keys can be found on the [Configuration Table](configuration-table.html) page.
+A complete list of configuration keys can be found on the [Configuration Table](configuration-table.html) page. Note
+that configuration keys prefixed with "sensitive." are treated specially, in that the values associated with such keys
+will be masked in logs and Samza's YARN ApplicationMaster UI. This is to prevent accidental disclosure only; no
+encryption is done.
## [Packaging »](packaging.html)
http://git-wip-us.apache.org/repos/asf/samza/blob/cad2c828/samza-api/src/main/java/org/apache/samza/config/Config.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/config/Config.java b/samza-api/src/main/java/org/apache/samza/config/Config.java
index 9f7ade0..2b99050 100644
--- a/samza-api/src/main/java/org/apache/samza/config/Config.java
+++ b/samza-api/src/main/java/org/apache/samza/config/Config.java
@@ -33,6 +33,9 @@ import java.util.regex.Pattern;
* Store and retrieve named, typed values as configuration for classes implementing this interface.
*/
public abstract class Config implements Map<String, String> {
+ public static final String SENSITIVE_PREFIX = "sensitive.";
+ public static final String SENSITIVE_MASK = "********";
+
public Config subset(String prefix) {
return subset(prefix, true);
}
@@ -221,6 +224,8 @@ public abstract class Config implements Map<String, String> {
}
}
+ public abstract Config sanitize();
+
public void clear() {
throw new ConfigException("Config is immutable.");
}
http://git-wip-us.apache.org/repos/asf/samza/blob/cad2c828/samza-api/src/main/java/org/apache/samza/config/MapConfig.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/config/MapConfig.java b/samza-api/src/main/java/org/apache/samza/config/MapConfig.java
index 1a83923..38d7424 100644
--- a/samza-api/src/main/java/org/apache/samza/config/MapConfig.java
+++ b/samza-api/src/main/java/org/apache/samza/config/MapConfig.java
@@ -31,11 +31,11 @@ import java.util.Set;
*/
public class MapConfig extends Config {
private final Map<String, String> map;
-
+
public MapConfig() {
this.map = Collections.emptyMap();
}
-
+
public MapConfig(Map<String, String> map) {
this(Collections.singletonList(map));
}
@@ -79,6 +79,21 @@ public class MapConfig extends Config {
}
@Override
+ public Config sanitize() {
+ return new MapConfig(sanitizeMap());
+ }
+
+ private Map<String, String> sanitizeMap() {
+ Map<String, String> sanitized = new HashMap<String, String>(map);
+ for (Entry<String, String> entry : sanitized.entrySet()) {
+ if (entry.getKey().startsWith(SENSITIVE_PREFIX)) {
+ entry.setValue(SENSITIVE_MASK);
+ }
+ }
+ return sanitized;
+ }
+
+ @Override
public int hashCode() {
final int prime = 31;
int result = 1;
@@ -105,6 +120,6 @@ public class MapConfig extends Config {
@Override
public String toString() {
- return map.toString();
+ return sanitizeMap().toString();
}
}
http://git-wip-us.apache.org/repos/asf/samza/blob/cad2c828/samza-api/src/test/java/org/apache/samza/config/TestConfig.java
----------------------------------------------------------------------
diff --git a/samza-api/src/test/java/org/apache/samza/config/TestConfig.java b/samza-api/src/test/java/org/apache/samza/config/TestConfig.java
index b4100c2..d9f378d 100644
--- a/samza-api/src/test/java/org/apache/samza/config/TestConfig.java
+++ b/samza-api/src/test/java/org/apache/samza/config/TestConfig.java
@@ -19,7 +19,7 @@
package org.apache.samza.config;
-import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.*;
import java.util.Map;
import java.util.HashMap;
@@ -61,4 +61,23 @@ public class TestConfig {
Class c4 = getClass(mc.getLong("testkey", defaultLong));
assertEquals(Long.class, c4);
}
+
+ @Test
+ public void testSanitize() {
+ Map<String, String> m = new HashMap<String, String>() {{
+ put("key1", "value1");
+ put("key2", "value2");
+ put("sensitive.key3", "secret1");
+ put("sensitive.key4", "secret2");
+ }};
+
+ Config config = new MapConfig(m);
+ assertFalse(config.toString().contains("secret"));
+
+ Config sanitized = config.sanitize();
+ assertEquals("value1", sanitized.get("key1"));
+ assertEquals("value2", sanitized.get("key2"));
+ assertEquals(Config.SENSITIVE_MASK, sanitized.get("sensitive.key3"));
+ assertEquals(Config.SENSITIVE_MASK, sanitized.get("sensitive.key4"));
+ }
}
http://git-wip-us.apache.org/repos/asf/samza/blob/cad2c828/samza-yarn/src/main/scala/org/apache/samza/webapp/ApplicationMasterRestServlet.scala
----------------------------------------------------------------------
diff --git a/samza-yarn/src/main/scala/org/apache/samza/webapp/ApplicationMasterRestServlet.scala b/samza-yarn/src/main/scala/org/apache/samza/webapp/ApplicationMasterRestServlet.scala
index 0afee64..4c855bf 100644
--- a/samza-yarn/src/main/scala/org/apache/samza/webapp/ApplicationMasterRestServlet.scala
+++ b/samza-yarn/src/main/scala/org/apache/samza/webapp/ApplicationMasterRestServlet.scala
@@ -103,6 +103,6 @@ class ApplicationMasterRestServlet(config: Config, state: SamzaAppMasterState, r
}
get("/config") {
- jsonMapper.writeValueAsString(new HashMap[String, Object](config.toMap))
+ jsonMapper.writeValueAsString(new HashMap[String, Object](config.sanitize.toMap))
}
}
http://git-wip-us.apache.org/repos/asf/samza/blob/cad2c828/samza-yarn/src/main/scala/org/apache/samza/webapp/ApplicationMasterWebServlet.scala
----------------------------------------------------------------------
diff --git a/samza-yarn/src/main/scala/org/apache/samza/webapp/ApplicationMasterWebServlet.scala b/samza-yarn/src/main/scala/org/apache/samza/webapp/ApplicationMasterWebServlet.scala
index 734d9d2..7fd5122 100644
--- a/samza-yarn/src/main/scala/org/apache/samza/webapp/ApplicationMasterWebServlet.scala
+++ b/samza-yarn/src/main/scala/org/apache/samza/webapp/ApplicationMasterWebServlet.scala
@@ -37,7 +37,7 @@ class ApplicationMasterWebServlet(config: Config, state: SamzaAppMasterState) ex
get("/") {
layoutTemplate("/WEB-INF/views/index.scaml",
- "config" -> TreeMap(config.toMap.toArray: _*),
+ "config" -> TreeMap(config.sanitize.toMap.toArray: _*),
"state" -> state,
"rmHttpAddress" -> WebAppUtils.getRMWebAppURLWithScheme(yarnConfig))
}
[26/36] samza git commit: SAMZA-613: update 0.9.0 release version.
bump up to 0.10.0-SNAPSHOT
Posted by cr...@apache.org.
SAMZA-613: update 0.9.0 release version. bump up to 0.10.0-SNAPSHOT
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/44d07f68
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/44d07f68
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/44d07f68
Branch: refs/heads/samza-sql
Commit: 44d07f681c423167eb320451d913c44d397ab8aa
Parents: 2c9020b
Author: Yan Fang <ya...@gmail.com>
Authored: Tue Mar 24 13:52:16 2015 -0700
Committer: Yan Fang <ya...@gmail.com>
Committed: Tue Mar 24 13:52:16 2015 -0700
----------------------------------------------------------------------
docs/_config.yml | 2 +-
gradle.properties | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/44d07f68/docs/_config.yml
----------------------------------------------------------------------
diff --git a/docs/_config.yml b/docs/_config.yml
index 4968a99..a3bb8d4 100644
--- a/docs/_config.yml
+++ b/docs/_config.yml
@@ -25,4 +25,4 @@ exclude: [_docs]
baseurl: http://samza.apache.org
version: latest
# this is the version you will go if you click 'switch version' in "latest" pages.
-latest-release: 0.8
+latest-release: 0.9
http://git-wip-us.apache.org/repos/asf/samza/blob/44d07f68/gradle.properties
----------------------------------------------------------------------
diff --git a/gradle.properties b/gradle.properties
index ed539e7..b18c0cb 100644
--- a/gradle.properties
+++ b/gradle.properties
@@ -15,7 +15,7 @@
# specific language governing permissions and limitations
# under the License.
group=org.apache.samza
-version=0.9.0-SNAPSHOT
+version=0.10.0-SNAPSHOT
scalaVersion=2.10
gradleVersion=2.0
[27/36] samza git commit: SAMZA-367: add Timer to the docs
Posted by cr...@apache.org.
SAMZA-367: add Timer to the docs
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/d20ae5c7
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/d20ae5c7
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/d20ae5c7
Branch: refs/heads/samza-sql
Commit: d20ae5c77170eee1951dc371b27ff14166aa2110
Parents: 44d07f6
Author: Yan Fang <ya...@gmail.com>
Authored: Wed Mar 25 17:11:50 2015 -0700
Committer: Yan Fang <ya...@gmail.com>
Committed: Wed Mar 25 17:11:50 2015 -0700
----------------------------------------------------------------------
docs/learn/documentation/versioned/container/metrics.md | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/d20ae5c7/docs/learn/documentation/versioned/container/metrics.md
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/container/metrics.md b/docs/learn/documentation/versioned/container/metrics.md
index 8ec7740..11a62f9 100644
--- a/docs/learn/documentation/versioned/container/metrics.md
+++ b/docs/learn/documentation/versioned/container/metrics.md
@@ -95,7 +95,7 @@ public class MyJavaStreamTask implements StreamTask, InitableTask {
}
{% endhighlight %}
-Samza currently supports two kind of metrics: [counters](../api/javadocs/org/apache/samza/metrics/Counter.html) and [gauges](../api/javadocs/org/apache/samza/metrics/Gauge.html). Use a counter when you want to track how often something occurs, and a gauge when you want to report the level of something, such as the size of a buffer. Each task instance (for each input stream partition) gets its own set of metrics.
+Samza currently supports three kinds of metrics: [counters](../api/javadocs/org/apache/samza/metrics/Counter.html), [gauges](../api/javadocs/org/apache/samza/metrics/Gauge.html) and [timer](../api/javadocs/org/apache/samza/metrics/Timer.html). Use a counter when you want to track how often something occurs, a gauge when you want to report the level of something, such as the size of a buffer, and a timer when you want to know how much time the block of code spends. Each task instance (for each input stream partition) gets its own set of metrics.
If you want to report metrics in some other way, e.g. directly to a graphing system (without going via Kafka), you can implement a [MetricsReporterFactory](../api/javadocs/org/apache/samza/metrics/MetricsReporterFactory.html) and reference it in your job configuration.
[04/36] samza git commit: SAMZA-505;
warn when using an array as a key in cached store
Posted by cr...@apache.org.
SAMZA-505; warn when using an array as a key in cached store
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/ff40b12e
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/ff40b12e
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/ff40b12e
Branch: refs/heads/samza-sql
Commit: ff40b12e36c0cc3fd8c65b75358be56df4967727
Parents: fcb5cea
Author: Chris Riccomini <cr...@apache.org>
Authored: Wed Mar 11 09:04:38 2015 -0700
Committer: Chris Riccomini <cr...@apache.org>
Committed: Wed Mar 11 09:04:38 2015 -0700
----------------------------------------------------------------------
build.gradle | 1 +
.../apache/samza/container/SamzaContainer.scala | 1 +
.../samza/serializers/ByteBufferSerde.scala | 49 ++++++++++++++++++++
.../apache/samza/serializers/ByteSerde.scala | 3 +-
.../samza/serializers/TestByteBufferSerde.scala | 41 ++++++++++++++++
.../apache/samza/storage/kv/CachedStore.scala | 15 ++++++
.../samza/storage/kv/TestCachedStore.scala | 35 ++++++++++++++
7 files changed, 144 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/ff40b12e/build.gradle
----------------------------------------------------------------------
diff --git a/build.gradle b/build.gradle
index 0a268ac..08583e0 100644
--- a/build.gradle
+++ b/build.gradle
@@ -317,6 +317,7 @@ project(":samza-kv_$scalaVersion") {
compile project(":samza-core_$scalaVersion")
compile "org.scala-lang:scala-library:$scalaLibVersion"
testCompile "junit:junit:$junitVersion"
+ testCompile "org.mockito:mockito-all:$mockitoVersion"
}
}
http://git-wip-us.apache.org/repos/asf/samza/blob/ff40b12e/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
index 275eb1a..87fcf58 100644
--- a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
+++ b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
@@ -121,6 +121,7 @@ object SamzaContainer extends Logging {
}
val serde = serdeName match {
case "byte" => getSerde(classOf[ByteSerdeFactory].getCanonicalName)
+ case "bytebuffer" => getSerde(classOf[ByteBufferSerdeFactory].getCanonicalName)
case "integer" => getSerde(classOf[IntegerSerdeFactory].getCanonicalName)
case "json" => getSerde(classOf[JsonSerdeFactory].getCanonicalName)
case "long" => getSerde(classOf[LongSerdeFactory].getCanonicalName)
http://git-wip-us.apache.org/repos/asf/samza/blob/ff40b12e/samza-core/src/main/scala/org/apache/samza/serializers/ByteBufferSerde.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/serializers/ByteBufferSerde.scala b/samza-core/src/main/scala/org/apache/samza/serializers/ByteBufferSerde.scala
new file mode 100644
index 0000000..05c3e38
--- /dev/null
+++ b/samza-core/src/main/scala/org/apache/samza/serializers/ByteBufferSerde.scala
@@ -0,0 +1,49 @@
+/*
+ * 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.samza.serializers
+
+import org.apache.samza.config.Config
+import java.nio.ByteBuffer
+
+/**
+ * A serializer for bytes that is effectively a no-op but can be useful for
+ * binary messages.
+ */
+class ByteBufferSerdeFactory extends SerdeFactory[ByteBuffer] {
+ def getSerde(name: String, config: Config): Serde[ByteBuffer] = new ByteBufferSerde
+}
+
+class ByteBufferSerde extends Serde[ByteBuffer] {
+ def toBytes(byteBuffer: ByteBuffer) = {
+ if (byteBuffer != null) {
+ val bytes = new Array[Byte](byteBuffer.remaining())
+ byteBuffer.get(bytes)
+ bytes
+ } else {
+ null
+ }
+ }
+
+ def fromBytes(bytes: Array[Byte]) = if (bytes != null) {
+ ByteBuffer.wrap(bytes)
+ } else {
+ null
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/ff40b12e/samza-core/src/main/scala/org/apache/samza/serializers/ByteSerde.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/serializers/ByteSerde.scala b/samza-core/src/main/scala/org/apache/samza/serializers/ByteSerde.scala
index e7ce09f..968da26 100644
--- a/samza-core/src/main/scala/org/apache/samza/serializers/ByteSerde.scala
+++ b/samza-core/src/main/scala/org/apache/samza/serializers/ByteSerde.scala
@@ -22,7 +22,8 @@ package org.apache.samza.serializers
import org.apache.samza.config.Config
/**
- * A serializer for bytes that is effectively a no-op but can be useful for binary messages.
+ * A serializer for bytes that is effectively a no-op but can be useful for
+ * binary messages.
*/
class ByteSerdeFactory extends SerdeFactory[Array[Byte]] {
def getSerde(name: String, config: Config): Serde[Array[Byte]] = new ByteSerde
http://git-wip-us.apache.org/repos/asf/samza/blob/ff40b12e/samza-core/src/test/scala/org/apache/samza/serializers/TestByteBufferSerde.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/serializers/TestByteBufferSerde.scala b/samza-core/src/test/scala/org/apache/samza/serializers/TestByteBufferSerde.scala
new file mode 100644
index 0000000..9401d70
--- /dev/null
+++ b/samza-core/src/test/scala/org/apache/samza/serializers/TestByteBufferSerde.scala
@@ -0,0 +1,41 @@
+/*
+ * 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.samza.serializers
+
+import java.util.Arrays
+import org.junit.Assert._
+import org.junit.Test
+import java.nio.ByteBuffer
+
+class TestByteBufferSerde {
+ @Test
+ def test {
+ val serde = new ByteBufferSerde
+ assertNull(serde.toBytes(null))
+ assertNull(serde.fromBytes(null))
+
+ val bytes = "A lazy way of creating a byte array".getBytes()
+ val testBytes = ByteBuffer.wrap(bytes)
+ testBytes.mark()
+ assertArrayEquals(serde.toBytes(testBytes), bytes)
+ testBytes.reset()
+ assertEquals(serde.fromBytes(bytes), testBytes)
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/samza/blob/ff40b12e/samza-kv/src/main/scala/org/apache/samza/storage/kv/CachedStore.scala
----------------------------------------------------------------------
diff --git a/samza-kv/src/main/scala/org/apache/samza/storage/kv/CachedStore.scala b/samza-kv/src/main/scala/org/apache/samza/storage/kv/CachedStore.scala
index 84cf6db..61bb3f6 100644
--- a/samza-kv/src/main/scala/org/apache/samza/storage/kv/CachedStore.scala
+++ b/samza-kv/src/main/scala/org/apache/samza/storage/kv/CachedStore.scala
@@ -72,6 +72,9 @@ class CachedStore[K, V](
}
}
+ /** tracks whether an array has been used as a key. since this is dangerous with LinkedHashMap, we want to warn on it. **/
+ private var containsArrayKeys = false
+
// Use counters here, rather than directly accessing variables using .size
// since metrics can be accessed in other threads, and cache.size is not
// thread safe since we're using a LinkedHashMap, and dirty.size is slow
@@ -109,6 +112,8 @@ class CachedStore[K, V](
def put(key: K, value: V) {
metrics.puts.inc
+ checkKeyIsArray(key)
+
// Add the key to the front of the dirty list (and remove any prior
// occurrences to dedupe).
val found = cache.get(key)
@@ -194,6 +199,16 @@ class CachedStore[K, V](
store.close
}
+
+ private def checkKeyIsArray(key: K) {
+ if (!containsArrayKeys && key.isInstanceOf[Array[_]]) {
+ // Warn the first time that we see an array key.
+ warn("Using arrays as keys results in unpredictable behavior since cache is implemented with a map. Consider using ByteBuffer, or a different key type.")
+ containsArrayKeys = true
+ }
+ }
+
+ def hasArrayKeys = containsArrayKeys
}
private case class CacheEntry[K, V](var value: V, var dirty: mutable.DoubleLinkedList[K])
http://git-wip-us.apache.org/repos/asf/samza/blob/ff40b12e/samza-kv/src/test/scala/org/apache/samza/storage/kv/TestCachedStore.scala
----------------------------------------------------------------------
diff --git a/samza-kv/src/test/scala/org/apache/samza/storage/kv/TestCachedStore.scala b/samza-kv/src/test/scala/org/apache/samza/storage/kv/TestCachedStore.scala
new file mode 100644
index 0000000..d03ec92
--- /dev/null
+++ b/samza-kv/src/test/scala/org/apache/samza/storage/kv/TestCachedStore.scala
@@ -0,0 +1,35 @@
+/*
+ * 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.samza.storage.kv
+
+import org.junit.Test
+import org.junit.Assert._
+import org.mockito.Mockito._
+
+class TestCachedStore {
+ @Test
+ def testArrayCheck {
+ val kv = mock(classOf[KeyValueStore[Array[Byte], Array[Byte]]])
+ val store = new CachedStore[Array[Byte], Array[Byte]](kv, 100, 100)
+ assertFalse(store.hasArrayKeys)
+ store.put("test1-key".getBytes("UTF-8"), "test1-value".getBytes("UTF-8"))
+ assertTrue(store.hasArrayKeys)
+ }
+}
\ No newline at end of file
[12/36] samza git commit: SAMZA-600: Move Java tests into
src/test/java in samza-kafka
Posted by cr...@apache.org.
SAMZA-600: Move Java tests into src/test/java in samza-kafka
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/660d879e
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/660d879e
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/660d879e
Branch: refs/heads/samza-sql
Commit: 660d879e7f1218a7289c97d654e8371fd7da6392
Parents: 1202df4
Author: Benjamin Fradet <benjamin dot fradet at gmail dot com>
Authored: Sat Mar 14 14:54:06 2015 -0700
Committer: Jakob Homan <jg...@gmail.com>
Committed: Sat Mar 14 14:54:06 2015 -0700
----------------------------------------------------------------------
.../samza/system/kafka/MockKafkaProducer.java | 246 +++++++++++++++++++
.../java/org/apache/samza/utils/TestUtils.java | 112 +++++++++
.../samza/system/kafka/MockKafkaProducer.java | 246 -------------------
.../scala/org/apache/samza/utils/TestUtils.java | 112 ---------
4 files changed, 358 insertions(+), 358 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/660d879e/samza-kafka/src/test/java/org/apache/samza/system/kafka/MockKafkaProducer.java
----------------------------------------------------------------------
diff --git a/samza-kafka/src/test/java/org/apache/samza/system/kafka/MockKafkaProducer.java b/samza-kafka/src/test/java/org/apache/samza/system/kafka/MockKafkaProducer.java
new file mode 100644
index 0000000..6f498de
--- /dev/null
+++ b/samza-kafka/src/test/java/org/apache/samza/system/kafka/MockKafkaProducer.java
@@ -0,0 +1,246 @@
+/*
+ * 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.samza.system.kafka;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReferenceArray;
+import org.apache.kafka.clients.producer.Callback;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Metric;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.samza.utils.TestUtils;
+import org.apache.kafka.common.MetricName;
+
+
+public class MockKafkaProducer implements Producer<byte[], byte[]> {
+
+ private Cluster _cluster;
+ private List<FutureTask<RecordMetadata>> _callbacksList = new ArrayList<FutureTask<RecordMetadata>>();
+ private boolean shouldBuffer = false;
+ private boolean errorNext = false;
+ private Exception exception = null;
+ private AtomicInteger msgsSent = new AtomicInteger(0);
+
+ /*
+ * Helps mock out buffered behavior seen in KafkaProducer. This MockKafkaProducer enables you to:
+ * - Create send that will instantly succeed & return a successful future
+ * - Set error for the next message that is sent (using errorNext). In this case, the next call to send returns a
+ * future with exception.
+ * Please note that errorNext is reset to false once a message send has failed. This means that errorNext has to be
+ * manually set to true in the unit test, before expecting failure for another message.
+ * - "shouldBuffer" can be turned on to start buffering messages. This will store all the callbacks and execute it
+ * at a later point of time in a separate thread. This thread NEEDS to be triggered from the unit test itself
+ * using "startDelayedSendThread" method
+ * - "Offset" in RecordMetadata is not guranteed to be correct
+ */
+ public MockKafkaProducer(int numNodes, String topicName, int numPartitions) {
+ this._cluster = TestUtils.clusterWith(numNodes, topicName, numPartitions);
+ }
+
+ public void setShouldBuffer(boolean shouldBuffer) {
+ this.shouldBuffer = shouldBuffer;
+ }
+
+ public void setErrorNext(boolean errorNext, Exception exception) {
+ this.errorNext = errorNext;
+ this.exception = exception;
+ }
+
+ public int getMsgsSent() {
+ return this.msgsSent.get();
+ }
+
+ public Thread startDelayedSendThread(final int sleepTime) {
+ Thread t = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ FutureTask[] callbackArray = new FutureTask[_callbacksList.size()];
+ AtomicReferenceArray<FutureTask> _bufferList = new AtomicReferenceArray<FutureTask>(_callbacksList.toArray(callbackArray));
+ ExecutorService executor = Executors.newFixedThreadPool(10);
+ try {
+ for(int i = 0; i < _bufferList.length(); i++) {
+ Thread.sleep(sleepTime);
+ FutureTask f = _bufferList.get(i);
+ if(!f.isDone()) {
+ executor.submit(f).get();
+ }
+ }
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ } catch (ExecutionException ee) {
+ ee.printStackTrace();
+ }
+ }
+ });
+ t.start();
+ return t;
+ }
+
+ @Override
+ public Future<RecordMetadata> send(ProducerRecord record) {
+ return send(record, null);
+ }
+
+ private RecordMetadata getRecordMetadata(ProducerRecord record) {
+ return new RecordMetadata(new TopicPartition(record.topic(), record.partition() == null ? 0 : record.partition()), 0, this.msgsSent.get());
+ }
+
+ @Override
+ public Future<RecordMetadata> send(final ProducerRecord record, final Callback callback) {
+ if (errorNext) {
+ if (shouldBuffer) {
+ FutureTask<RecordMetadata> f = new FutureTask<RecordMetadata>(new Callable<RecordMetadata>() {
+ @Override
+ public RecordMetadata call()
+ throws Exception {
+ callback.onCompletion(null, exception);
+ return getRecordMetadata(record);
+ }
+ });
+ _callbacksList.add(f);
+ this.errorNext = false;
+ return f;
+ } else {
+ callback.onCompletion(null, this.exception);
+ this.errorNext = false;
+ return new FutureFailure(this.exception);
+ }
+ } else {
+ if (shouldBuffer) {
+ FutureTask<RecordMetadata> f = new FutureTask<RecordMetadata>(new Callable<RecordMetadata>() {
+ @Override
+ public RecordMetadata call()
+ throws Exception {
+ msgsSent.incrementAndGet();
+ RecordMetadata metadata = getRecordMetadata(record);
+ callback.onCompletion(metadata, null);
+ return metadata;
+ }
+ });
+ _callbacksList.add(f);
+ return f;
+ } else {
+ int offset = msgsSent.incrementAndGet();
+ final RecordMetadata metadata = getRecordMetadata(record);
+ callback.onCompletion(metadata, null);
+ return new FutureSuccess(record, offset);
+ }
+ }
+ }
+
+ @Override
+ public List<PartitionInfo> partitionsFor(String topic) {
+ return this._cluster.partitionsForTopic(topic);
+ }
+
+ @Override
+ public Map<MetricName, Metric> metrics() {
+ return null;
+ }
+
+ @Override
+ public void close() {
+
+ }
+
+ private static class FutureFailure implements Future<RecordMetadata> {
+
+ private final ExecutionException exception;
+
+ public FutureFailure(Exception exception) {
+ this.exception = new ExecutionException(exception);
+ }
+
+ @Override
+ public boolean cancel(boolean interrupt) {
+ return false;
+ }
+
+ @Override
+ public RecordMetadata get() throws ExecutionException {
+ throw this.exception;
+ }
+
+ @Override
+ public RecordMetadata get(long timeout, TimeUnit unit) throws ExecutionException {
+ throw this.exception;
+ }
+
+ @Override
+ public boolean isCancelled() {
+ return false;
+ }
+
+ @Override
+ public boolean isDone() {
+ return true;
+ }
+ }
+
+ private static class FutureSuccess implements Future<RecordMetadata> {
+
+ private ProducerRecord record;
+ private final RecordMetadata _metadata;
+
+ public FutureSuccess(ProducerRecord record, int offset) {
+ this.record = record;
+ this._metadata = new RecordMetadata(new TopicPartition(record.topic(), record.partition() == null ? 0 : record.partition()), 0, offset);
+ }
+
+ @Override
+ public boolean cancel(boolean interrupt) {
+ return false;
+ }
+
+ @Override
+ public RecordMetadata get() throws ExecutionException {
+ return this._metadata;
+ }
+
+ @Override
+ public RecordMetadata get(long timeout, TimeUnit unit) throws ExecutionException {
+ return this._metadata;
+ }
+
+ @Override
+ public boolean isCancelled() {
+ return false;
+ }
+
+ @Override
+ public boolean isDone() {
+ return true;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/660d879e/samza-kafka/src/test/java/org/apache/samza/utils/TestUtils.java
----------------------------------------------------------------------
diff --git a/samza-kafka/src/test/java/org/apache/samza/utils/TestUtils.java b/samza-kafka/src/test/java/org/apache/samza/utils/TestUtils.java
new file mode 100644
index 0000000..2fa743f
--- /dev/null
+++ b/samza-kafka/src/test/java/org/apache/samza/utils/TestUtils.java
@@ -0,0 +1,112 @@
+/**
+ * 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.samza.utils;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+
+import static java.util.Arrays.asList;
+
+
+/**
+ * Copied from :kafka-clients API as a workaround until KAFKA-1861 is resolved
+ * Helper functions for writing unit tests
+ */
+public class TestUtils {
+
+ public static File IO_TMP_DIR = new File(System.getProperty("java.io.tmpdir"));
+
+ public static String LETTERS = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz";
+ public static String DIGITS = "0123456789";
+ public static String LETTERS_AND_DIGITS = LETTERS + DIGITS;
+
+ /* A consistent random number generator to make tests repeatable */
+ public static final Random seededRandom = new Random(192348092834L);
+ public static final Random random = new Random();
+
+ public static Cluster singletonCluster(String topic, int partitions) {
+ return clusterWith(1, topic, partitions);
+ }
+
+ public static Cluster clusterWith(int nodes, String topic, int partitions) {
+ Node[] ns = new Node[nodes];
+ for (int i = 0; i < nodes; i++)
+ ns[i] = new Node(0, "localhost", 1969);
+ List<PartitionInfo> parts = new ArrayList<PartitionInfo>();
+ for (int i = 0; i < partitions; i++)
+ parts.add(new PartitionInfo(topic, i, ns[i % ns.length], ns, ns));
+ return new Cluster(asList(ns), parts);
+ }
+
+ /**
+ * Choose a number of random available ports
+ */
+ public static int[] choosePorts(int count) {
+ try {
+ ServerSocket[] sockets = new ServerSocket[count];
+ int[] ports = new int[count];
+ for (int i = 0; i < count; i++) {
+ sockets[i] = new ServerSocket(0);
+ ports[i] = sockets[i].getLocalPort();
+ }
+ for (int i = 0; i < count; i++)
+ sockets[i].close();
+ return ports;
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Choose an available port
+ */
+ public static int choosePort() {
+ return choosePorts(1)[0];
+ }
+
+ /**
+ * Generate an array of random bytes
+ *
+ * @param size The size of the array
+ */
+ public static byte[] randomBytes(int size) {
+ byte[] bytes = new byte[size];
+ seededRandom.nextBytes(bytes);
+ return bytes;
+ }
+
+ /**
+ * Generate a random string of letters and digits of the given length
+ *
+ * @param len The length of the string
+ * @return The random string
+ */
+ public static String randomString(int len) {
+ StringBuilder b = new StringBuilder();
+ for (int i = 0; i < len; i++)
+ b.append(LETTERS_AND_DIGITS.charAt(seededRandom.nextInt(LETTERS_AND_DIGITS.length())));
+ return b.toString();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/660d879e/samza-kafka/src/test/scala/org/apache/samza/system/kafka/MockKafkaProducer.java
----------------------------------------------------------------------
diff --git a/samza-kafka/src/test/scala/org/apache/samza/system/kafka/MockKafkaProducer.java b/samza-kafka/src/test/scala/org/apache/samza/system/kafka/MockKafkaProducer.java
deleted file mode 100644
index 6f498de..0000000
--- a/samza-kafka/src/test/scala/org/apache/samza/system/kafka/MockKafkaProducer.java
+++ /dev/null
@@ -1,246 +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.samza.system.kafka;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.FutureTask;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicReferenceArray;
-import org.apache.kafka.clients.producer.Callback;
-import org.apache.kafka.clients.producer.Producer;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.clients.producer.RecordMetadata;
-import org.apache.kafka.common.Cluster;
-import org.apache.kafka.common.Metric;
-import org.apache.kafka.common.PartitionInfo;
-import org.apache.kafka.common.TopicPartition;
-import org.apache.samza.utils.TestUtils;
-import org.apache.kafka.common.MetricName;
-
-
-public class MockKafkaProducer implements Producer<byte[], byte[]> {
-
- private Cluster _cluster;
- private List<FutureTask<RecordMetadata>> _callbacksList = new ArrayList<FutureTask<RecordMetadata>>();
- private boolean shouldBuffer = false;
- private boolean errorNext = false;
- private Exception exception = null;
- private AtomicInteger msgsSent = new AtomicInteger(0);
-
- /*
- * Helps mock out buffered behavior seen in KafkaProducer. This MockKafkaProducer enables you to:
- * - Create send that will instantly succeed & return a successful future
- * - Set error for the next message that is sent (using errorNext). In this case, the next call to send returns a
- * future with exception.
- * Please note that errorNext is reset to false once a message send has failed. This means that errorNext has to be
- * manually set to true in the unit test, before expecting failure for another message.
- * - "shouldBuffer" can be turned on to start buffering messages. This will store all the callbacks and execute it
- * at a later point of time in a separate thread. This thread NEEDS to be triggered from the unit test itself
- * using "startDelayedSendThread" method
- * - "Offset" in RecordMetadata is not guranteed to be correct
- */
- public MockKafkaProducer(int numNodes, String topicName, int numPartitions) {
- this._cluster = TestUtils.clusterWith(numNodes, topicName, numPartitions);
- }
-
- public void setShouldBuffer(boolean shouldBuffer) {
- this.shouldBuffer = shouldBuffer;
- }
-
- public void setErrorNext(boolean errorNext, Exception exception) {
- this.errorNext = errorNext;
- this.exception = exception;
- }
-
- public int getMsgsSent() {
- return this.msgsSent.get();
- }
-
- public Thread startDelayedSendThread(final int sleepTime) {
- Thread t = new Thread(new Runnable() {
- @Override
- public void run() {
- FutureTask[] callbackArray = new FutureTask[_callbacksList.size()];
- AtomicReferenceArray<FutureTask> _bufferList = new AtomicReferenceArray<FutureTask>(_callbacksList.toArray(callbackArray));
- ExecutorService executor = Executors.newFixedThreadPool(10);
- try {
- for(int i = 0; i < _bufferList.length(); i++) {
- Thread.sleep(sleepTime);
- FutureTask f = _bufferList.get(i);
- if(!f.isDone()) {
- executor.submit(f).get();
- }
- }
- } catch (InterruptedException e) {
- e.printStackTrace();
- } catch (ExecutionException ee) {
- ee.printStackTrace();
- }
- }
- });
- t.start();
- return t;
- }
-
- @Override
- public Future<RecordMetadata> send(ProducerRecord record) {
- return send(record, null);
- }
-
- private RecordMetadata getRecordMetadata(ProducerRecord record) {
- return new RecordMetadata(new TopicPartition(record.topic(), record.partition() == null ? 0 : record.partition()), 0, this.msgsSent.get());
- }
-
- @Override
- public Future<RecordMetadata> send(final ProducerRecord record, final Callback callback) {
- if (errorNext) {
- if (shouldBuffer) {
- FutureTask<RecordMetadata> f = new FutureTask<RecordMetadata>(new Callable<RecordMetadata>() {
- @Override
- public RecordMetadata call()
- throws Exception {
- callback.onCompletion(null, exception);
- return getRecordMetadata(record);
- }
- });
- _callbacksList.add(f);
- this.errorNext = false;
- return f;
- } else {
- callback.onCompletion(null, this.exception);
- this.errorNext = false;
- return new FutureFailure(this.exception);
- }
- } else {
- if (shouldBuffer) {
- FutureTask<RecordMetadata> f = new FutureTask<RecordMetadata>(new Callable<RecordMetadata>() {
- @Override
- public RecordMetadata call()
- throws Exception {
- msgsSent.incrementAndGet();
- RecordMetadata metadata = getRecordMetadata(record);
- callback.onCompletion(metadata, null);
- return metadata;
- }
- });
- _callbacksList.add(f);
- return f;
- } else {
- int offset = msgsSent.incrementAndGet();
- final RecordMetadata metadata = getRecordMetadata(record);
- callback.onCompletion(metadata, null);
- return new FutureSuccess(record, offset);
- }
- }
- }
-
- @Override
- public List<PartitionInfo> partitionsFor(String topic) {
- return this._cluster.partitionsForTopic(topic);
- }
-
- @Override
- public Map<MetricName, Metric> metrics() {
- return null;
- }
-
- @Override
- public void close() {
-
- }
-
- private static class FutureFailure implements Future<RecordMetadata> {
-
- private final ExecutionException exception;
-
- public FutureFailure(Exception exception) {
- this.exception = new ExecutionException(exception);
- }
-
- @Override
- public boolean cancel(boolean interrupt) {
- return false;
- }
-
- @Override
- public RecordMetadata get() throws ExecutionException {
- throw this.exception;
- }
-
- @Override
- public RecordMetadata get(long timeout, TimeUnit unit) throws ExecutionException {
- throw this.exception;
- }
-
- @Override
- public boolean isCancelled() {
- return false;
- }
-
- @Override
- public boolean isDone() {
- return true;
- }
- }
-
- private static class FutureSuccess implements Future<RecordMetadata> {
-
- private ProducerRecord record;
- private final RecordMetadata _metadata;
-
- public FutureSuccess(ProducerRecord record, int offset) {
- this.record = record;
- this._metadata = new RecordMetadata(new TopicPartition(record.topic(), record.partition() == null ? 0 : record.partition()), 0, offset);
- }
-
- @Override
- public boolean cancel(boolean interrupt) {
- return false;
- }
-
- @Override
- public RecordMetadata get() throws ExecutionException {
- return this._metadata;
- }
-
- @Override
- public RecordMetadata get(long timeout, TimeUnit unit) throws ExecutionException {
- return this._metadata;
- }
-
- @Override
- public boolean isCancelled() {
- return false;
- }
-
- @Override
- public boolean isDone() {
- return true;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/660d879e/samza-kafka/src/test/scala/org/apache/samza/utils/TestUtils.java
----------------------------------------------------------------------
diff --git a/samza-kafka/src/test/scala/org/apache/samza/utils/TestUtils.java b/samza-kafka/src/test/scala/org/apache/samza/utils/TestUtils.java
deleted file mode 100644
index 2fa743f..0000000
--- a/samza-kafka/src/test/scala/org/apache/samza/utils/TestUtils.java
+++ /dev/null
@@ -1,112 +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.samza.utils;
-
-import java.io.File;
-import java.io.IOException;
-import java.net.ServerSocket;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Random;
-import org.apache.kafka.common.Cluster;
-import org.apache.kafka.common.Node;
-import org.apache.kafka.common.PartitionInfo;
-
-import static java.util.Arrays.asList;
-
-
-/**
- * Copied from :kafka-clients API as a workaround until KAFKA-1861 is resolved
- * Helper functions for writing unit tests
- */
-public class TestUtils {
-
- public static File IO_TMP_DIR = new File(System.getProperty("java.io.tmpdir"));
-
- public static String LETTERS = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz";
- public static String DIGITS = "0123456789";
- public static String LETTERS_AND_DIGITS = LETTERS + DIGITS;
-
- /* A consistent random number generator to make tests repeatable */
- public static final Random seededRandom = new Random(192348092834L);
- public static final Random random = new Random();
-
- public static Cluster singletonCluster(String topic, int partitions) {
- return clusterWith(1, topic, partitions);
- }
-
- public static Cluster clusterWith(int nodes, String topic, int partitions) {
- Node[] ns = new Node[nodes];
- for (int i = 0; i < nodes; i++)
- ns[i] = new Node(0, "localhost", 1969);
- List<PartitionInfo> parts = new ArrayList<PartitionInfo>();
- for (int i = 0; i < partitions; i++)
- parts.add(new PartitionInfo(topic, i, ns[i % ns.length], ns, ns));
- return new Cluster(asList(ns), parts);
- }
-
- /**
- * Choose a number of random available ports
- */
- public static int[] choosePorts(int count) {
- try {
- ServerSocket[] sockets = new ServerSocket[count];
- int[] ports = new int[count];
- for (int i = 0; i < count; i++) {
- sockets[i] = new ServerSocket(0);
- ports[i] = sockets[i].getLocalPort();
- }
- for (int i = 0; i < count; i++)
- sockets[i].close();
- return ports;
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- /**
- * Choose an available port
- */
- public static int choosePort() {
- return choosePorts(1)[0];
- }
-
- /**
- * Generate an array of random bytes
- *
- * @param size The size of the array
- */
- public static byte[] randomBytes(int size) {
- byte[] bytes = new byte[size];
- seededRandom.nextBytes(bytes);
- return bytes;
- }
-
- /**
- * Generate a random string of letters and digits of the given length
- *
- * @param len The length of the string
- * @return The random string
- */
- public static String randomString(int len) {
- StringBuilder b = new StringBuilder();
- for (int i = 0; i < len; i++)
- b.append(LETTERS_AND_DIGITS.charAt(seededRandom.nextInt(LETTERS_AND_DIGITS.length())));
- return b.toString();
- }
-
-}
[19/36] samza git commit: SAMZA-567;
allow stores to work from init method
Posted by cr...@apache.org.
SAMZA-567; allow stores to work from init method
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/9a9ab8d2
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/9a9ab8d2
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/9a9ab8d2
Branch: refs/heads/samza-sql
Commit: 9a9ab8d243a940701cf8b6374b15d507a484f491
Parents: a6ada1f
Author: Naveen Somasundaram <na...@gmail.com>
Authored: Wed Mar 18 18:50:46 2015 -0700
Committer: Chris Riccomini <cr...@apache.org>
Committed: Wed Mar 18 18:50:46 2015 -0700
----------------------------------------------------------------------
.../main/scala/org/apache/samza/container/SamzaContainer.scala | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/9a9ab8d2/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
index 87fcf58..9fc3b55 100644
--- a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
+++ b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
@@ -541,8 +541,8 @@ class SamzaContainer(
startMetrics
startOffsetManager
startStores
- startTask
startProducers
+ startTask
startConsumers
info("Entering run loop.")
@@ -555,8 +555,8 @@ class SamzaContainer(
info("Shutting down.")
shutdownConsumers
- shutdownProducers
shutdownTask
+ shutdownProducers
shutdownStores
shutdownOffsetManager
shutdownMetrics
[23/36] samza git commit: SAMZA-611: added Yan Fang's public key
Posted by cr...@apache.org.
SAMZA-611: added Yan Fang's public key
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/6f5bafb6
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/6f5bafb6
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/6f5bafb6
Branch: refs/heads/samza-sql
Commit: 6f5bafb6cd93934781161eb6b1868d11ea347c95
Parents: 28e2dbb
Author: Yan Fang <ya...@gmail.com>
Authored: Tue Mar 24 11:05:11 2015 -0700
Committer: Yan Fang <ya...@gmail.com>
Committed: Tue Mar 24 11:05:11 2015 -0700
----------------------------------------------------------------------
KEYS | 60 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-
1 file changed, 59 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/6f5bafb6/KEYS
----------------------------------------------------------------------
diff --git a/KEYS b/KEYS
index 993961c..e5dc701 100644
--- a/KEYS
+++ b/KEYS
@@ -26,7 +26,7 @@ Developers:
Releases will be signed using one of these keys in this file. This file will
be available with the distributed Apache Samza releases at:
- https://dist.apache.org/repos/dist/release/samza/KEYS
+ https://dist.apache.org/repos/dist/samza/KEYS
********************************************************************************
@@ -212,3 +212,61 @@ ji5kmDimbMpgui6X+ekFGEvA6O7sqTugVz2gLqZz2U9ZLO+FShkV4s775KVnzXwO
oIhD2jZh+lFFojHtt+TXM+l3u9lfjjP7
=yVrO
-----END PGP PUBLIC KEY BLOCK-----
+pub 4096R/EA00BA80 2015-03-24 [expires: 2020-03-22]
+uid Yan Fang (CODE SIGNING KEY) <ya...@apache.org>
+sig 3 EA00BA80 2015-03-24 Yan Fang (CODE SIGNING KEY) <ya...@apache.org>
+sub 4096R/E3F3DAD3 2015-03-24 [expires: 2020-03-22]
+sig EA00BA80 2015-03-24 Yan Fang (CODE SIGNING KEY) <ya...@apache.org>
+
+-----BEGIN PGP PUBLIC KEY BLOCK-----
+Version: GnuPG/MacGPG2 v2.0.20 (Darwin)
+
+mQINBFURDNwBEADMV6g16KWRCHcCCknGxVTBImc9rMyR5O0Rigwi8as0EIXNLuRr
+AgxTRo6J+y5x0qKn90q2LJ5FHOge7n7+EEy1PCx79x6YXMK5K2VpuWecrka+J5p7
+XBX2A7sXLvOqokPuzM8yqKcCnD+lcG3BncTcmbvyXGSDsamUdGXAbdUN5dqQjdkn
+Cm3fYRI9cpJIZavLBj/yRegWifQO/HmuUs/9VYNILBRkIroQcXILyFb8kiEzHNZe
+nbHmDSb6CIVipbycrL20NgPjVEkJqqc6qytmWFXvz9gv/s3J2OhtV024LeANPKNL
+tJv1JSKaLGfOZ9l8R0jo5v5y2LnbGrbK89A3SBre7IDeZA55e8PDWtDJ3n/6168U
+Ys6nJMjkVFp0V2SnjHzr0VF2MQ4622rpHEMuvsP2LMMQ+2n1EOVvpi1OX5mxkgRa
+tjD3V+TB+VYVF4AAF3oSKplGiZ2mhjcpykYyzj6159u7cpPaBKV3i663qjeAmXs5
+ArFniiwxkImjtgWXWcPVcofv+56Wg7DsVwvJCBAm3ZhyEquuwF4K5HrKUJHCtLhV
+x6AoA/SE3PBfdMWzaDMb5Z7b7+rbUzdkLaoF2T5IcvxmhIZR7DUTPOwz8kKO/rAl
+2lwOd5uS9srrYywMPkZ2Mx5JrdVGOpXMaynDYh9Hls/xFXH8y2e+UTeYHwARAQAB
+tDBZYW4gRmFuZyAoQ09ERSBTSUdOSU5HIEtFWSkgPHlhbmZhbmdAYXBhY2hlLm9y
+Zz6JAj0EEwECACcCGwMFCQlmAYACHgECF4AFAlURDU0FCwkIBwMFFQoJCAsFFgID
+AQAACgkQysBiOeoAuoAceQ/+LYQKU4vIZLgMGwj0J5x3pVNkTuLfO8BR2d+FNmN1
+DE2O5ex5CW/ZicgqQ3HIsTl7DG4Rb7rocdFB2vajrBjzpu0Zd4B5Xrm7b+8DxCaN
+79PMVGzkgPt9Mqt3YAo9/NNyq/+1lOqoKOlJe7tcm+tC0v3YwQhf1jdlaG9iFNnB
+PnBwc8geA6ueo6CfO5ORRVVcdkvO7mKI1hjH7T45DciBih8awLRU8a1z8+q2LhY7
+s9V5RmmegGxhYRlE16cxfgqU5ldtYwIYWu8t8mFEA3SoXQxxau/b1Rq50H06j9L7
+e3uB/MWHxv7OhjfaQU3qs3eXxbHg7TQ7cupSJhP0e+ikyensdn80JyHO8coTbgrI
+qBFPRBOWjmsOSfCZ673/9D8GeFOXH8bAfPLL6RB//TIoS04RshHC5bonxUMjHAnL
+aU9nAXAIak4HdeFvcnb3Owzw/1QP97TYU3g2FWkIqPJ9dDAad/RGWnu+iGNBAiRT
+oJJ22iuWsVYT4vdyN9Awj8EELmOdLmEeaAiEBjnanFFAi3b44hfKsEo+UlM7Mxzq
+xikxM8SGOuQ584FA0Tj/0QFKklpTqQBQiW8Z8Dj5UbjiOu6H2XJ6IMnSBBH6HheY
+GtaNXInYve0MH34wIa/IcgSU5fOOtyrIv6ReS0wXu6oReEijYBWMkyht1G7GfgFo
+Xge5Ag0EVREM3AEQAMtNIThvSsWI6o8cyOOGPkvQnyK0UJJBmaDApX6+kukmVKNC
+yODaoZHbKcJC9ACCd5Y4A6EPr8bqHhuExj/1l3CV9ZKJuscq3loOz4kUDtrcSOrq
+bxZJVEYGo69a6nT7TeCIU3hrem6EmuROoXlxerCcTAvKucGYIgqiAtneax/SYI/y
+uwhyJLLiv1UEknG1/Dd49/hGmXhkKR31iakWZGGoyCNNV2YTw53L27Q/CGA/oYR6
+pMKdlwna4as5a3hYqdy3ABqJohwBo7pgrTCE67qi0sy8wcUMbkOfqS7pwC0+E6jG
+2fcZ4dhat+/T63IvIOifLUO7oQXstaCVEXO2uBwA57yYdrD1u4QhQHavQg2vZis1
+Pp3CNLVxKaGkO33TPQ0CXVWfxAQBdm0PSZBStGIwCTs27nCH5zNmHKR7tjwZUmkE
+qjM4ToYy0QMUdTravwbeOcwBe/GgU9BjOPJlcmACpAMitj+fatPjB3lAtdcGG8eU
+BAkAaVDXmRpRQsaf3/9Q+rrZEy6pr3SP9D5uwvcqq+L8EytqpaMItOo/IAj/KNQF
+nDDSz8i4sJAOgJT7TiPDkNHGSFMR4N0FmyEYosRs6DVk7iJ7RUFfinb+VYbnmMw4
+4kJbjQk77aj/WcKastK+Xh/V3nPAHJ+R09FSB2n5wNe+9VkL82BAzkuRekilABEB
+AAGJAiUEGAECAA8FAlURDNwCGwwFCQlmAYAACgkQysBiOeoAuoD9ww//frVaAyGO
++eAnFEPm2/NqG5Ws31ktUzD3BmdSQmaM1/A0l9Gm+Q4EDPT5apNe+tad9l06bSPt
+9Kx92k2zI2Iz4NojIBmcf3kWli4lcgwhqYiLI/NwBmTBCRg4C7DdPfp6JxdPhij+
+n8oYROoF/ZTArghIUHd8mYcgiuQ93c/HlgqBloG31e9D2VduR0DMP35wMol5kRyx
+lhaSVb40/ZIKJmd3m3efN9s81B2DGKkxhVgz3n3KuW3/nqWomT0+XIIUcl8hHxsV
+lHKTgecGqsQulZYSzgrIeGqTZPSMzsmHxELsrlRvgLTaeXRFI7s88Y1uQuhEw8xb
+1kxVUraNO+8Mex/SFcLWMc2FK3V9L5UKflA2Dr0s4xjSvJTS4AVUHmHbpSb+3MOz
+X2RMnwxr0zjNxZzRZhcNASV1s6+QuQ3W9wYwZdty0Zu8egACzn3WvQZ7dQNp7Fta
+svzDY7mCF3QC6GjRHpsSn34CaIODkJXKizblffAY1TV5W8ruVJbQFTh7X5UpzfKL
+hWVDj45vqXr+R6VUOV7wY0VGs9V9C9xxBjuAvy4Ko7PDSO6gsEzZj/4QToeNZvE3
+NLXrrWu2oFxKIbpdAEDFWT8CJN+qHMOB/w+iw9RDwpPDakMBTrV/j8mWXKnRyoAy
+hG+A4+kpvCAxGqau99KDqxvq6XzMx6prN9s=
+=h097
+-----END PGP PUBLIC KEY BLOCK-----
[07/36] samza git commit: SAMZA-550;
remove dead code from serializer config
Posted by cr...@apache.org.
SAMZA-550; remove dead code from serializer config
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/13ee7062
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/13ee7062
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/13ee7062
Branch: refs/heads/samza-sql
Commit: 13ee706225678172675163f088f499f78474ce43
Parents: c725b3c
Author: Benjamin Fradet <be...@gmail.com>
Authored: Thu Mar 12 13:35:05 2015 -0700
Committer: Chris Riccomini <cr...@apache.org>
Committed: Thu Mar 12 13:35:05 2015 -0700
----------------------------------------------------------------------
.../src/main/scala/org/apache/samza/config/SerializerConfig.scala | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/13ee7062/samza-core/src/main/scala/org/apache/samza/config/SerializerConfig.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/config/SerializerConfig.scala b/samza-core/src/main/scala/org/apache/samza/config/SerializerConfig.scala
index 0c026e6..8e3eb57 100644
--- a/samza-core/src/main/scala/org/apache/samza/config/SerializerConfig.scala
+++ b/samza-core/src/main/scala/org/apache/samza/config/SerializerConfig.scala
@@ -35,8 +35,9 @@ class SerializerConfig(config: Config) extends ScalaMapConfig(config) {
* Returns a list of all serializer names from the config file. Useful for
* getting individual serializers.
*/
+ import SerializerConfig._
def getSerdeNames() = {
- val subConf = config.subset("serializers.registry.", true)
+ val subConf = config.subset(SERIALIZER_PREFIX format "", true)
subConf.keys.filter(k => k.endsWith(".class")).map(_.replace(".class", ""))
}
}
[34/36] samza git commit: SAMZA-624: Remove Travis integration
Posted by cr...@apache.org.
SAMZA-624: Remove Travis integration
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/b276afec
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/b276afec
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/b276afec
Branch: refs/heads/samza-sql
Commit: b276afecfad7b1d542b2a5737471c783bb4ac079
Parents: 457a524
Author: Jakob Homan <jg...@gmail.com>
Authored: Tue Apr 7 14:19:18 2015 -0700
Committer: Jakob Homan <jg...@gmail.com>
Committed: Tue Apr 7 14:53:20 2015 -0700
----------------------------------------------------------------------
.travis.yml | 46 -------------------------------------------
README.md | 2 --
docs/contribute/tests.md | 6 ------
3 files changed, 54 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/b276afec/.travis.yml
----------------------------------------------------------------------
diff --git a/.travis.yml b/.travis.yml
deleted file mode 100644
index 33bfa1c..0000000
--- a/.travis.yml
+++ /dev/null
@@ -1,46 +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.
-#
-
-language: java
-
-jdk:
- - oraclejdk7
- - oraclejdk8
-
-env:
- matrix:
- - SCALA_VERSION='-PscalaVersion=2.10' YARN_VERSION='-PyarnVersion=2.4.0'
- - SCALA_VERSION='-PscalaVersion=2.10' YARN_VERSION='-PyarnVersion=2.5.0'
-
-matrix:
-
-script: ./gradlew $SCALA_VERSION $YARN_VERSION --stacktrace check
-
-# no install since gradlew check does all the dependency downloads as part of the check command
-install: true
-
-notifications:
- irc:
- channels:
- - "irc.freenode.org#samza"
- on_success: always
- on_failure: always
- template:
- - "%{repository} (%{branch}:%{commit} by %{author}): %{message} (%{build_url})"
- email:
- - dev@samza.apache.org
http://git-wip-us.apache.org/repos/asf/samza/blob/b276afec/README.md
----------------------------------------------------------------------
diff --git a/README.md b/README.md
index a6ae16f..7f92020 100644
--- a/README.md
+++ b/README.md
@@ -34,8 +34,6 @@ Samza builds with [Scala](http://www.scala-lang.org/) 2.10 and [YARN](http://had
### Testing Samza
-[](https://travis-ci.org/apache/samza)
-
To run all tests:
./gradlew clean test
http://git-wip-us.apache.org/repos/asf/samza/blob/b276afec/docs/contribute/tests.md
----------------------------------------------------------------------
diff --git a/docs/contribute/tests.md b/docs/contribute/tests.md
index 0171bf9..bda13d0 100644
--- a/docs/contribute/tests.md
+++ b/docs/contribute/tests.md
@@ -62,12 +62,6 @@ On Mac, check-all.sh will default to the appropriate path for each environment v
JAVA8_HOME is not set.
JAVA8_HOME defaulted to /Library/Java/JavaVirtualMachines/jdk1.8.0_20.jdk/Contents/Home
-### Travis CI
-
-[Travis CI](https://travis-ci.org/apache/incubator-samza) has been configured to run Samza's unit tests after every commit to Samza's [master branch](https://git-wip-us.apache.org/repos/asf?p=samza.git;a=tree). The test results are mailed to the [developer mailing list](/community/mailing-lists.html), and posted in the [IRC channel](/community/irc.html).
-
-[](https://travis-ci.org/apache/incubator-samza)
-
### Running Integration Tests Locally
Samza uses [Zopkio](https://github.com/linkedin/Zopkio) to deploy and execute its integration tests. Integration tests can be executed by running:
[08/36] samza git commit: SAMZA-596; upgrade scala test to 2.2.4
Posted by cr...@apache.org.
SAMZA-596; upgrade scala test to 2.2.4
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/2904bc9b
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/2904bc9b
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/2904bc9b
Branch: refs/heads/samza-sql
Commit: 2904bc9be9d3525617d97f1ca32d49722aedc4c6
Parents: 13ee706
Author: Bennett Andrews <be...@gmail.com>
Authored: Thu Mar 12 18:44:58 2015 -0700
Committer: Chris Riccomini <cr...@apache.org>
Committed: Thu Mar 12 18:44:58 2015 -0700
----------------------------------------------------------------------
gradle/dependency-versions.gradle | 2 +-
.../src/test/scala/org/apache/samza/container/TestRunLoop.scala | 4 ++--
.../scala/org/apache/samza/system/TestStreamMetadataCache.scala | 4 ++--
3 files changed, 5 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/2904bc9b/gradle/dependency-versions.gradle
----------------------------------------------------------------------
diff --git a/gradle/dependency-versions.gradle b/gradle/dependency-versions.gradle
index 84be50b..cb6ef21 100644
--- a/gradle/dependency-versions.gradle
+++ b/gradle/dependency-versions.gradle
@@ -22,7 +22,7 @@
jacksonVersion = "1.8.5"
junitVersion = "4.8.1"
mockitoVersion = "1.8.4"
- scalaTestVersion = "1.9.2"
+ scalaTestVersion = "2.2.4"
zkClientVersion = "0.3"
zookeeperVersion = "3.3.4"
metricsVersion = "2.2.0"
http://git-wip-us.apache.org/repos/asf/samza/blob/2904bc9b/samza-core/src/test/scala/org/apache/samza/container/TestRunLoop.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/container/TestRunLoop.scala b/samza-core/src/test/scala/org/apache/samza/container/TestRunLoop.scala
index fb1ebdd..2a0897f 100644
--- a/samza-core/src/test/scala/org/apache/samza/container/TestRunLoop.scala
+++ b/samza-core/src/test/scala/org/apache/samza/container/TestRunLoop.scala
@@ -25,14 +25,14 @@ import org.mockito.Mockito._
import org.mockito.invocation.InvocationOnMock
import org.mockito.stubbing.Answer
import org.scalatest.junit.AssertionsForJUnit
-import org.scalatest.matchers.ShouldMatchers
+import org.scalatest.{Matchers => ScalaTestMatchers}
import org.scalatest.mock.MockitoSugar
import org.apache.samza.Partition
import org.apache.samza.system.{ IncomingMessageEnvelope, SystemConsumers, SystemStreamPartition }
import org.apache.samza.task.ReadableCoordinator
import org.apache.samza.task.TaskCoordinator.RequestScope
-class TestRunLoop extends AssertionsForJUnit with MockitoSugar with ShouldMatchers {
+class TestRunLoop extends AssertionsForJUnit with MockitoSugar with ScalaTestMatchers {
class StopRunLoop extends RuntimeException
val p0 = new Partition(0)
http://git-wip-us.apache.org/repos/asf/samza/blob/2904bc9b/samza-core/src/test/scala/org/apache/samza/system/TestStreamMetadataCache.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/system/TestStreamMetadataCache.scala b/samza-core/src/test/scala/org/apache/samza/system/TestStreamMetadataCache.scala
index 89ea078..4716d97 100644
--- a/samza-core/src/test/scala/org/apache/samza/system/TestStreamMetadataCache.scala
+++ b/samza-core/src/test/scala/org/apache/samza/system/TestStreamMetadataCache.scala
@@ -24,12 +24,12 @@ import org.apache.samza.system.SystemStreamMetadata.SystemStreamPartitionMetadat
import org.apache.samza.util.Clock
import org.junit.{Before, Test}
import org.mockito.Mockito._
-import org.scalatest.matchers.ShouldMatchers
+import org.scalatest.{Matchers => ScalaTestMatchers}
import org.scalatest.junit.AssertionsForJUnit
import org.scalatest.mock.MockitoSugar
import scala.collection.JavaConversions._
-class TestStreamMetadataCache extends AssertionsForJUnit with MockitoSugar with ShouldMatchers {
+class TestStreamMetadataCache extends AssertionsForJUnit with MockitoSugar with ScalaTestMatchers {
def makeMetadata(streamNames: Set[String] = Set("stream"), numPartitions: Int = 4) = {
val partitions = (0 until numPartitions).map(partition => {
new Partition(partition) -> new SystemStreamPartitionMetadata("oldest", "newest", "upcoming")
[24/36] samza git commit: SAMZA-612: remove "incubating" in
release.gradle
Posted by cr...@apache.org.
SAMZA-612: remove "incubating" in release.gradle
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/91289f49
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/91289f49
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/91289f49
Branch: refs/heads/samza-sql
Commit: 91289f491b0e5aa1b0b12b5872181e29438d7350
Parents: 6f5bafb
Author: Yan Fang <ya...@gmail.com>
Authored: Tue Mar 24 11:06:12 2015 -0700
Committer: Yan Fang <ya...@gmail.com>
Committed: Tue Mar 24 11:06:12 2015 -0700
----------------------------------------------------------------------
gradle/release.gradle | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/91289f49/gradle/release.gradle
----------------------------------------------------------------------
diff --git a/gradle/release.gradle b/gradle/release.gradle
index 2baca7f..e501098 100644
--- a/gradle/release.gradle
+++ b/gradle/release.gradle
@@ -27,7 +27,7 @@ task sourceRelease(type: Tar) {
description = "Build a source release, specifically excluding the build directories and gradle wrapper files"
compression = Compression.GZIP
- baseName = "apache-samza-${project.version}-incubating-src"
+ baseName = "apache-samza-${project.version}-src"
from(project.rootDir) {
exclude '**/build'
@@ -68,7 +68,7 @@ subprojects {
modifyPom {
project {
- name 'Apache Samza (incubating)'
+ name 'Apache Samza'
description 'A distributed stream processing framework built upon Apache Kafka and Apache Hadoop YARN.'
url 'http://samza.apache.org/'
[09/36] samza git commit: SAMZA-576;
fork a new process for every test in samza-kafka to try and fix
TestKafkaSystemAdmin.testShouldGetOldestNewestAndNextOffsets
Posted by cr...@apache.org.
SAMZA-576; fork a new process for every test in samza-kafka to try and fix TestKafkaSystemAdmin.testShouldGetOldestNewestAndNextOffsets
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/769844e5
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/769844e5
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/769844e5
Branch: refs/heads/samza-sql
Commit: 769844e533b157b504a8459f059dd70ff47fc1eb
Parents: 2904bc9
Author: Chris Riccomini <cr...@apache.org>
Authored: Fri Mar 13 11:30:36 2015 -0700
Committer: Chris Riccomini <cr...@apache.org>
Committed: Fri Mar 13 11:30:36 2015 -0700
----------------------------------------------------------------------
build.gradle | 5 +++++
1 file changed, 5 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/769844e5/build.gradle
----------------------------------------------------------------------
diff --git a/build.gradle b/build.gradle
index 08583e0..516c83e 100644
--- a/build.gradle
+++ b/build.gradle
@@ -175,6 +175,11 @@ project(":samza-kafka_$scalaVersion") {
minHeapSize = "1560m"
maxHeapSize = "1560m"
jvmArgs = ["-XX:+UseConcMarkSweepGC", "-server"]
+ // There appear to be issues between TestKafkaSystemAdmin and
+ // TestKafkaCheckpointManager both running brokeres and ZK. Restarting the
+ // gradle worker after every test clears things up. These tests should be
+ // moved to the integration test suite.
+ forkEvery = 1
}
}
[17/36] samza git commit: SAMZA-605;
fix jobrunner package name in docs
Posted by cr...@apache.org.
SAMZA-605; fix jobrunner package name in docs
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/137d0d67
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/137d0d67
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/137d0d67
Branch: refs/heads/samza-sql
Commit: 137d0d67e03e7ec598efa8c181d05c358104dd6b
Parents: 537262e
Author: Benjamin Fradet <be...@gmail.com>
Authored: Tue Mar 17 15:26:23 2015 -0700
Committer: Chris Riccomini <cr...@apache.org>
Committed: Tue Mar 17 15:26:23 2015 -0700
----------------------------------------------------------------------
docs/learn/documentation/versioned/jobs/job-runner.md | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/137d0d67/docs/learn/documentation/versioned/jobs/job-runner.md
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/jobs/job-runner.md b/docs/learn/documentation/versioned/jobs/job-runner.md
index c2f6b09..5f0607f 100644
--- a/docs/learn/documentation/versioned/jobs/job-runner.md
+++ b/docs/learn/documentation/versioned/jobs/job-runner.md
@@ -23,7 +23,7 @@ Samza jobs are started using a script called run-job.sh.
{% highlight bash %}
samza-example/target/bin/run-job.sh \
- --config-factory=samza.config.factories.PropertiesConfigFactory \
+ --config-factory=org.apache.samza.config.factories.PropertiesConfigFactory \
--config-path=file://$PWD/config/hello-world.properties
{% endhighlight %}
[02/36] samza git commit: SAMZA-455;
post jdk support policy to the website
Posted by cr...@apache.org.
SAMZA-455; post jdk support policy to the website
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/7a21beff
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/7a21beff
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/7a21beff
Branch: refs/heads/samza-sql
Commit: 7a21beff807a40a40457225cbef249fb31b5e7b5
Parents: 57075c4
Author: Chris Riccomini <cr...@apache.org>
Authored: Mon Mar 2 12:26:36 2015 -0800
Committer: Chris Riccomini <cr...@apache.org>
Committed: Mon Mar 2 12:26:36 2015 -0800
----------------------------------------------------------------------
docs/startup/download/index.md | 4 ++++
1 file changed, 4 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/7a21beff/docs/startup/download/index.md
----------------------------------------------------------------------
diff --git a/docs/startup/download/index.md b/docs/startup/download/index.md
index 329969e..47ef0e0 100644
--- a/docs/startup/download/index.md
+++ b/docs/startup/download/index.md
@@ -23,6 +23,10 @@ Samza is released as a source artifact, and also through Maven.
If you just want to play around with Samza for the first time, go to [Hello Samza](/startup/hello-samza/{{site.version}}).
+### JDK Notice
+
+Starting in June of 2015, Samza will begin requiring JDK7 or higher. Please see [this mailing list thread](http://mail-archives.apache.org/mod_mbox/samza-dev/201502.mbox/%3CCABYbY7eGFKgSgqAPot1qC8LjKu%3DKGYOLcNg3WuxfYzySAS-d1Q%40mail.gmail.com%3E) for details on this decision.
+
### Source Releases
* [samza-sources-0.8.0-incubating.tgz](http://www.apache.org/dyn/closer.cgi/samza/0.8.0-incubating)
[21/36] samza git commit: SAMZA-586;
disable compressed messages when sending to a store's changelog topic
from a kafka system
Posted by cr...@apache.org.
SAMZA-586; disable compressed messages when sending to a store's changelog topic from a kafka system
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/8b52e8ae
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/8b52e8ae
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/8b52e8ae
Branch: refs/heads/samza-sql
Commit: 8b52e8aecbca2915d50f0ae3600ceec799fac60f
Parents: 05a02ee
Author: Chris Riccomini <cr...@apache.org>
Authored: Thu Mar 19 09:25:29 2015 -0700
Committer: Chris Riccomini <cr...@apache.org>
Committed: Thu Mar 19 09:25:29 2015 -0700
----------------------------------------------------------------------
.../org/apache/samza/config/StorageConfig.scala | 16 +++++++-
.../apache/samza/config/TestStorageConfig.scala | 40 ++++++++++++++++++++
.../kafka/KafkaCheckpointManagerFactory.scala | 4 +-
.../samza/system/kafka/KafkaSystemFactory.scala | 12 +++++-
.../system/kafka/TestKafkaSystemFactory.scala | 13 +++++++
5 files changed, 81 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/8b52e8ae/samza-core/src/main/scala/org/apache/samza/config/StorageConfig.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/config/StorageConfig.scala b/samza-core/src/main/scala/org/apache/samza/config/StorageConfig.scala
index f977b8b..be3f106 100644
--- a/samza-core/src/main/scala/org/apache/samza/config/StorageConfig.scala
+++ b/samza-core/src/main/scala/org/apache/samza/config/StorageConfig.scala
@@ -20,8 +20,8 @@
package org.apache.samza.config
import scala.collection.JavaConversions._
-
import org.apache.samza.util.Logging
+import org.apache.samza.util.Util
object StorageConfig {
// stream config constants
@@ -43,4 +43,18 @@ class StorageConfig(config: Config) extends ScalaMapConfig(config) with Logging
val conf = config.subset("stores.", true)
conf.keys.filter(k => k.endsWith(".factory")).map(k => k.substring(0, k.length - ".factory".length)).toSeq
}
+
+ /**
+ * Helper method to check if a system has a changelog attached to it.
+ */
+ def isChangelogSystem(systemName: String) = {
+ config
+ .getStoreNames
+ // Get changelogs for all stores in the format of "system.stream"
+ .map(getChangelogStream(_))
+ .filter(_.isDefined)
+ // Convert "system.stream" to systemName
+ .map(systemStreamName => Util.getSystemStreamFromNames(systemStreamName.get).getSystem)
+ .contains(systemName)
+ }
}
http://git-wip-us.apache.org/repos/asf/samza/blob/8b52e8ae/samza-core/src/test/scala/org/apache/samza/config/TestStorageConfig.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/config/TestStorageConfig.scala b/samza-core/src/test/scala/org/apache/samza/config/TestStorageConfig.scala
new file mode 100644
index 0000000..81a35ec
--- /dev/null
+++ b/samza-core/src/test/scala/org/apache/samza/config/TestStorageConfig.scala
@@ -0,0 +1,40 @@
+/*
+ * 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.samza.config
+
+import scala.collection.JavaConversions._
+import org.apache.samza.config.StorageConfig._
+import org.junit.Assert.assertFalse
+import org.junit.Assert.assertTrue
+import org.junit.Test
+
+class TestStorageConfig {
+ @Test
+ def testIsChangelogSystem {
+ val configMap = Map[String, String](
+ FACTORY.format("system1") -> "some.factory.Class",
+ CHANGELOG_STREAM.format("system1") -> "system1.stream1",
+ FACTORY.format("system2") -> "some.factory.Class")
+ val config = new MapConfig(configMap)
+ assertFalse(config.isChangelogSystem("system3"))
+ assertFalse(config.isChangelogSystem("system2"))
+ assertTrue(config.isChangelogSystem("system1"))
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/samza/blob/8b52e8ae/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManagerFactory.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManagerFactory.scala b/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManagerFactory.scala
index 7fc6d89..3dfa26a 100644
--- a/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManagerFactory.scala
+++ b/samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManagerFactory.scala
@@ -43,8 +43,8 @@ object KafkaCheckpointManagerFactory {
val INJECTED_PRODUCER_PROPERTIES = Map(
"acks" -> "all",
// Forcibly disable compression because Kafka doesn't support compression
- // on log compacted topics. Details in SAMZA-393.
- "compression.codec" -> "none")
+ // on log compacted topics. Details in SAMZA-586.
+ "compression.type" -> "none")
// Set the checkpoint topic configs to have a very small segment size and
// enable log compaction. This keeps job startup time small since there
http://git-wip-us.apache.org/repos/asf/samza/blob/8b52e8ae/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemFactory.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemFactory.scala b/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemFactory.scala
index 4f15002..c84ceb7 100644
--- a/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemFactory.scala
+++ b/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemFactory.scala
@@ -25,9 +25,18 @@ import org.apache.samza.metrics.MetricsRegistry
import org.apache.samza.config.KafkaConfig.Config2Kafka
import org.apache.kafka.clients.producer.KafkaProducer
import org.apache.samza.system.SystemFactory
+import org.apache.samza.config.StorageConfig._
import org.I0Itec.zkclient.ZkClient
import kafka.utils.ZKStringSerializer
+object KafkaSystemFactory extends Logging {
+ def getInjectedProducerProperties(systemName: String, config: Config) = if (config.isChangelogSystem(systemName)) {
+ warn("System name '%s' is being used as a changelog. Disabling compression since Kafka does not support compression for log compacted topics." format systemName)
+ Map[String, String]("compression.type" -> "none")
+ } else {
+ Map[String, String]()
+ }
+}
class KafkaSystemFactory extends SystemFactory with Logging {
def getConsumer(systemName: String, config: Config, registry: MetricsRegistry) = {
@@ -66,7 +75,8 @@ class KafkaSystemFactory extends SystemFactory with Logging {
def getProducer(systemName: String, config: Config, registry: MetricsRegistry) = {
val clientId = KafkaUtil.getClientId("samza-producer", config)
- val producerConfig = config.getKafkaSystemProducerConfig(systemName, clientId)
+ val injectedProps = KafkaSystemFactory.getInjectedProducerProperties(systemName, config)
+ val producerConfig = config.getKafkaSystemProducerConfig(systemName, clientId, injectedProps)
val getProducer = () => { new KafkaProducer[Array[Byte], Array[Byte]](producerConfig.getProducerProperties) }
val metrics = new KafkaSystemProducerMetrics(systemName, registry)
http://git-wip-us.apache.org/repos/asf/samza/blob/8b52e8ae/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemFactory.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemFactory.scala b/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemFactory.scala
index 5f65144..ce84b6d 100644
--- a/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemFactory.scala
+++ b/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemFactory.scala
@@ -21,6 +21,7 @@ package org.apache.samza.system.kafka
import org.apache.samza.SamzaException
import org.apache.samza.config.MapConfig
+import org.apache.samza.config.StorageConfig
import org.apache.samza.metrics.MetricsRegistryMap
import org.apache.samza.system.SystemStream
import org.junit.Assert._
@@ -83,4 +84,16 @@ class TestKafkaSystemFactory {
assertNotNull(producer)
assertTrue(producer.isInstanceOf[KafkaSystemProducer])
}
+
+ @Test
+ def testInjectedProducerProps {
+ val configMap = Map[String, String](
+ StorageConfig.FACTORY.format("system1") -> "some.factory.Class",
+ StorageConfig.CHANGELOG_STREAM.format("system1") -> "system1.stream1",
+ StorageConfig.FACTORY.format("system2") -> "some.factory.Class")
+ val config = new MapConfig(configMap)
+ assertEquals(Map[String, String](), KafkaSystemFactory.getInjectedProducerProperties("system3", config))
+ assertEquals(Map[String, String](), KafkaSystemFactory.getInjectedProducerProperties("system2", config))
+ assertEquals(Map[String, String]("compression.type" -> "none"), KafkaSystemFactory.getInjectedProducerProperties("system1", config))
+ }
}
[11/36] samza git commit: SAMZA-592;
ignore replica not available exceptions, and be more aggressive about
refreshing kafka topic metadata
Posted by cr...@apache.org.
SAMZA-592; ignore replica not available exceptions, and be more aggressive about refreshing kafka topic metadata
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/1202df41
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/1202df41
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/1202df41
Branch: refs/heads/samza-sql
Commit: 1202df4171f44a877997301218fe3c3037abf2d5
Parents: 23c4b39
Author: Chris Riccomini <cr...@apache.org>
Authored: Fri Mar 13 14:52:03 2015 -0700
Committer: Chris Riccomini <cr...@apache.org>
Committed: Fri Mar 13 14:52:03 2015 -0700
----------------------------------------------------------------------
.../kafka/KafkaCheckpointManager.scala | 7 ++--
.../apache/samza/system/kafka/BrokerProxy.scala | 3 +-
.../apache/samza/system/kafka/GetOffset.scala | 7 ++--
.../samza/system/kafka/KafkaSystemAdmin.scala | 10 ++---
.../samza/system/kafka/TopicMetadataCache.scala | 12 +++++-
.../scala/org/apache/samza/util/KafkaUtil.scala | 39 ++++++++++++++++----
.../system/kafka/TestKafkaSystemAdmin.scala | 8 ++--
.../system/kafka/TestTopicMetadataCache.scala | 14 ++++++-
.../org/apache/samza/utils/TestKafkaUtil.scala | 33 +++++++++++++++++
.../test/integration/TestStatefulTask.scala | 8 +---
10 files changed, 107 insertions(+), 34 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/1202df41/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 4a1b31f..c9504ec 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
@@ -44,6 +44,7 @@ import org.apache.samza.util.TopicMetadataStore
import scala.collection.mutable
import java.util.Properties
import org.apache.kafka.clients.producer.{Producer, ProducerRecord}
+import org.apache.samza.util.KafkaUtil
/**
* Kafka checkpoint manager is used to store checkpoints in a Kafka topic.
@@ -159,7 +160,7 @@ class KafkaCheckpointManager(
.get(topicAndPartition)
.getOrElse(throw new KafkaCheckpointException("Unable to find offset information for %s:0" format checkpointTopic))
// Fail or retry if there was an an issue with the offset request.
- ErrorMapping.maybeThrowException(offsetResponse.error)
+ KafkaUtil.maybeThrowException(offsetResponse.error)
val offset: Long = offsetResponse
.offsets
@@ -287,7 +288,7 @@ class KafkaCheckpointManager(
warn("Got an offset out of range exception while getting last entry in %s for topic %s and partition 0, so returning a null offset to the KafkaConsumer. Let it decide what to do based on its autooffset.reset setting." format (entryType, checkpointTopic))
return
}
- ErrorMapping.maybeThrowException(errorCode)
+ KafkaUtil.maybeThrowException(errorCode)
}
for (response <- fetchResponse.messageSet(checkpointTopic, 0)) {
@@ -385,7 +386,7 @@ class KafkaCheckpointManager(
loop => {
val topicMetadataMap = TopicMetadataCache.getTopicMetadata(Set(checkpointTopic), systemName, metadataStore.getTopicInfo)
val topicMetadata = topicMetadataMap(checkpointTopic)
- ErrorMapping.maybeThrowException(topicMetadata.errorCode)
+ KafkaUtil.maybeThrowException(topicMetadata.errorCode)
val partitionCount = topicMetadata.partitionsMetadata.length
if (partitionCount != 1) {
http://git-wip-us.apache.org/repos/asf/samza/blob/1202df41/samza-kafka/src/main/scala/org/apache/samza/system/kafka/BrokerProxy.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/scala/org/apache/samza/system/kafka/BrokerProxy.scala b/samza-kafka/src/main/scala/org/apache/samza/system/kafka/BrokerProxy.scala
index c6e231a..614f33f 100644
--- a/samza-kafka/src/main/scala/org/apache/samza/system/kafka/BrokerProxy.scala
+++ b/samza-kafka/src/main/scala/org/apache/samza/system/kafka/BrokerProxy.scala
@@ -34,6 +34,7 @@ import org.apache.samza.util.ThreadNamePrefix.SAMZA_THREAD_NAME_PREFIX
import scala.collection.JavaConversions._
import scala.collection.concurrent
import scala.collection.mutable
+import org.apache.samza.util.KafkaUtil
/**
* Companion object for class JvmMetrics encapsulating various constants
@@ -235,7 +236,7 @@ class BrokerProxy(
// handle the recoverable errors.
remainingErrors.foreach(e => {
warn("Got non-recoverable error codes during multifetch. Throwing an exception to trigger reconnect. Errors: %s" format remainingErrors.mkString(","))
- ErrorMapping.maybeThrowException(e.code) })
+ KafkaUtil.maybeThrowException(e.code) })
notLeaderOrUnknownTopic.foreach(e => abdicate(e.tp))
http://git-wip-us.apache.org/repos/asf/samza/blob/1202df41/samza-kafka/src/main/scala/org/apache/samza/system/kafka/GetOffset.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/scala/org/apache/samza/system/kafka/GetOffset.scala b/samza-kafka/src/main/scala/org/apache/samza/system/kafka/GetOffset.scala
index 147aabc..5528702 100644
--- a/samza-kafka/src/main/scala/org/apache/samza/system/kafka/GetOffset.scala
+++ b/samza-kafka/src/main/scala/org/apache/samza/system/kafka/GetOffset.scala
@@ -21,12 +21,13 @@
package org.apache.samza.system.kafka
-import kafka.common.{ OffsetOutOfRangeException, ErrorMapping }
+import kafka.common.OffsetOutOfRangeException
import kafka.api._
import kafka.common.TopicAndPartition
import kafka.api.PartitionOffsetRequestInfo
import org.apache.samza.util.Logging
import kafka.message.MessageAndOffset
+import org.apache.samza.util.KafkaUtil
/**
* GetOffset validates offsets for topic partitions, and manages fetching new
@@ -59,7 +60,7 @@ class GetOffset(
val messages = consumer.defaultFetch((topicAndPartition, offset.toLong))
if (messages.hasError) {
- ErrorMapping.maybeThrowException(messages.errorCode(topicAndPartition.topic, topicAndPartition.partition))
+ KafkaUtil.maybeThrowException(messages.errorCode(topicAndPartition.topic, topicAndPartition.partition))
}
info("Able to successfully read from offset %s for topic and partition %s. Using it to instantiate consumer." format (offset, topicAndPartition))
@@ -85,7 +86,7 @@ class GetOffset(
.get(topicAndPartition)
.getOrElse(toss("Unable to find offset information for %s" format topicAndPartition))
- ErrorMapping.maybeThrowException(partitionOffsetResponse.error)
+ KafkaUtil.maybeThrowException(partitionOffsetResponse.error)
partitionOffsetResponse
.offsets
http://git-wip-us.apache.org/repos/asf/samza/blob/1202df41/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemAdmin.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemAdmin.scala b/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemAdmin.scala
index b790be1..f783c57 100644
--- a/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemAdmin.scala
+++ b/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemAdmin.scala
@@ -28,12 +28,13 @@ import org.apache.samza.system.SystemStreamPartition
import org.apache.samza.util.{ClientUtilTopicMetadataStore, ExponentialSleepStrategy, Logging}
import kafka.api._
import kafka.consumer.SimpleConsumer
-import kafka.common.{TopicExistsException, TopicAndPartition, ErrorMapping}
+import kafka.common.{TopicExistsException, TopicAndPartition}
import java.util.{Properties, UUID}
import scala.collection.JavaConversions._
import org.apache.samza.system.SystemStreamMetadata.SystemStreamPartitionMetadata
import kafka.consumer.ConsumerConfig
import kafka.admin.AdminUtils
+import org.apache.samza.util.KafkaUtil
object KafkaSystemAdmin extends Logging {
/**
@@ -222,12 +223,11 @@ class KafkaSystemAdmin(
.values
// Convert the topic metadata to a Seq[(Broker, TopicAndPartition)]
.flatMap(topicMetadata => {
- ErrorMapping.maybeThrowException(topicMetadata.errorCode)
+ KafkaUtil.maybeThrowException(topicMetadata.errorCode)
topicMetadata
.partitionsMetadata
// Convert Seq[PartitionMetadata] to Seq[(Broker, TopicAndPartition)]
.map(partitionMetadata => {
- ErrorMapping.maybeThrowException(partitionMetadata.errorCode)
val topicAndPartition = new TopicAndPartition(topicMetadata.topic, partitionMetadata.partitionId)
val leader = partitionMetadata
.leader
@@ -263,7 +263,7 @@ class KafkaSystemAdmin(
.getOffsetsBefore(new OffsetRequest(partitionOffsetInfo))
.partitionErrorAndOffsets
.mapValues(partitionErrorAndOffset => {
- ErrorMapping.maybeThrowException(partitionErrorAndOffset.error)
+ KafkaUtil.maybeThrowException(partitionErrorAndOffset.error)
partitionErrorAndOffset.offsets.head
})
@@ -320,7 +320,7 @@ class KafkaSystemAdmin(
val metadataStore = new ClientUtilTopicMetadataStore(brokerListString, clientId, timeout)
val topicMetadataMap = TopicMetadataCache.getTopicMetadata(Set(topicName), systemName, metadataStore.getTopicInfo)
val topicMetadata = topicMetadataMap(topicName)
- ErrorMapping.maybeThrowException(topicMetadata.errorCode)
+ KafkaUtil.maybeThrowException(topicMetadata.errorCode)
val partitionCount = topicMetadata.partitionsMetadata.length
if (partitionCount < numKafkaChangelogPartitions) {
http://git-wip-us.apache.org/repos/asf/samza/blob/1202df41/samza-kafka/src/main/scala/org/apache/samza/system/kafka/TopicMetadataCache.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/scala/org/apache/samza/system/kafka/TopicMetadataCache.scala b/samza-kafka/src/main/scala/org/apache/samza/system/kafka/TopicMetadataCache.scala
index 4a49d22..82ecf1a 100644
--- a/samza-kafka/src/main/scala/org/apache/samza/system/kafka/TopicMetadataCache.scala
+++ b/samza-kafka/src/main/scala/org/apache/samza/system/kafka/TopicMetadataCache.scala
@@ -21,7 +21,7 @@ package org.apache.samza.system.kafka
import org.apache.samza.util.Logging
import kafka.api.TopicMetadata
-import kafka.common.ErrorMapping
+import org.apache.samza.util.KafkaUtil
/**
* TopicMetadataCache is used to cache all the topic metadata for Kafka per
@@ -43,7 +43,7 @@ object TopicMetadataCache extends Logging {
val missingTopics = topics.filter(topic => !topicMetadataMap.contains(systemName, topic))
val topicsWithBadOrExpiredMetadata = (topics -- missingTopics).filter(topic => {
val metadata = topicMetadataMap(systemName, topic)
- metadata.streamMetadata.errorCode != ErrorMapping.NoError || ((time - metadata.lastRefreshMs) > cacheTimeout)
+ hasBadErrorCode(metadata.streamMetadata) || ((time - metadata.lastRefreshMs) > cacheTimeout)
})
val topicsToRefresh = missingTopics ++ topicsWithBadOrExpiredMetadata
@@ -67,4 +67,12 @@ object TopicMetadataCache extends Logging {
def clear {
topicMetadataMap.clear
}
+
+ /**
+ * Helper method to check if a topic's metadata has a bad errorCode, or if a
+ * partition's metadata has a bad errorCode.
+ */
+ def hasBadErrorCode(streamMetadata: TopicMetadata) = {
+ KafkaUtil.isBadErrorCode(streamMetadata.errorCode) || streamMetadata.partitionsMetadata.exists(partitionMetadata => KafkaUtil.isBadErrorCode(partitionMetadata.errorCode))
+ }
}
http://git-wip-us.apache.org/repos/asf/samza/blob/1202df41/samza-kafka/src/main/scala/org/apache/samza/util/KafkaUtil.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/scala/org/apache/samza/util/KafkaUtil.scala b/samza-kafka/src/main/scala/org/apache/samza/util/KafkaUtil.scala
index 2482f23..a7a095b 100644
--- a/samza-kafka/src/main/scala/org/apache/samza/util/KafkaUtil.scala
+++ b/samza-kafka/src/main/scala/org/apache/samza/util/KafkaUtil.scala
@@ -19,15 +19,16 @@
package org.apache.samza.util
-import org.apache.samza.config.{Config, ConfigException}
+import java.util.concurrent.atomic.AtomicLong
+import org.apache.kafka.common.PartitionInfo
+import org.apache.samza.config.Config
+import org.apache.samza.config.ConfigException
import org.apache.samza.config.JobConfig.Config2Job
-import java.util.concurrent.atomic.{AtomicInteger, AtomicLong}
import org.apache.samza.system.OutgoingMessageEnvelope
-import org.apache.kafka.common.utils.Utils
-import java.util.Random
-import org.apache.kafka.common.PartitionInfo
+import kafka.common.ErrorMapping
+import kafka.common.ReplicaNotAvailableException
-object KafkaUtil {
+object KafkaUtil extends Logging {
val counter = new AtomicLong(0)
def getClientId(id: String, config: Config): String = getClientId(
@@ -43,10 +44,34 @@ object KafkaUtil {
System.currentTimeMillis,
counter.getAndIncrement)
- private def abs(n: Int) = if(n == Integer.MIN_VALUE) 0 else math.abs(n)
+ private def abs(n: Int) = if (n == Integer.MIN_VALUE) 0 else math.abs(n)
def getIntegerPartitionKey(envelope: OutgoingMessageEnvelope, partitions: java.util.List[PartitionInfo]): Integer = {
val numPartitions = partitions.size
abs(envelope.getPartitionKey.hashCode()) % numPartitions
}
+
+ /**
+ * Exactly the same as Kafka's ErrorMapping.maybeThrowException
+ * implementation, except suppresses ReplicaNotAvailableException exceptions.
+ * According to the Kafka
+ * <a href="https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol">protocol
+ * docs</a>, ReplicaNotAvailableException can be safely ignored.
+ */
+ def maybeThrowException(code: Short) {
+ try {
+ ErrorMapping.maybeThrowException(code)
+ } catch {
+ case e: ReplicaNotAvailableException =>
+ debug("Got ReplicaNotAvailableException, but ignoring since it's safe to do so.")
+ }
+ }
+
+ /**
+ * Checks if a Kafka errorCode is "bad" or not. "Bad" is defined as any
+ * errorCode that's not NoError and also not ReplicaNotAvailableCode.
+ */
+ def isBadErrorCode(code: Short) = {
+ code != ErrorMapping.NoError && code != ErrorMapping.ReplicaNotAvailableCode
+ }
}
http://git-wip-us.apache.org/repos/asf/samza/blob/1202df41/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemAdmin.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemAdmin.scala b/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemAdmin.scala
index 3d1e6ec..0380d35 100644
--- a/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemAdmin.scala
+++ b/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemAdmin.scala
@@ -22,9 +22,7 @@
package org.apache.samza.system.kafka
import java.util.Properties
-
import kafka.admin.AdminUtils
-import kafka.common.ErrorMapping
import kafka.consumer.Consumer
import kafka.consumer.ConsumerConfig
import kafka.server.KafkaConfig
@@ -34,7 +32,6 @@ import kafka.utils.TestZKUtils
import kafka.utils.Utils
import kafka.utils.ZKStringSerializer
import kafka.zk.EmbeddedZookeeper
-
import org.I0Itec.zkclient.ZkClient
import org.apache.samza.Partition
import org.apache.samza.system.SystemStreamMetadata
@@ -45,11 +42,12 @@ import org.apache.samza.util.ClientUtilTopicMetadataStore
import org.apache.samza.util.TopicMetadataStore
import org.junit.Assert._
import org.junit.{Test, BeforeClass, AfterClass}
-
import scala.collection.JavaConversions._
import org.apache.samza.config.KafkaProducerConfig
import org.apache.kafka.clients.producer.{ProducerRecord, KafkaProducer}
import java.util
+import kafka.common.ErrorMapping
+import org.apache.samza.util.KafkaUtil
object TestKafkaSystemAdmin {
val TOPIC = "input"
@@ -113,7 +111,7 @@ object TestKafkaSystemAdmin {
val topicMetadata = topicMetadataMap(TOPIC)
val errorCode = topicMetadata.errorCode
- ErrorMapping.maybeThrowException(errorCode)
+ KafkaUtil.maybeThrowException(errorCode)
done = expectedPartitionCount == topicMetadata.partitionsMetadata.size
} catch {
http://git-wip-us.apache.org/repos/asf/samza/blob/1202df41/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestTopicMetadataCache.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestTopicMetadataCache.scala b/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestTopicMetadataCache.scala
index e698d2f..50c13ab 100644
--- a/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestTopicMetadataCache.scala
+++ b/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestTopicMetadataCache.scala
@@ -21,15 +21,15 @@ package org.apache.samza.system.kafka
import java.util.concurrent.CountDownLatch
import java.util.concurrent.atomic.{ AtomicBoolean, AtomicInteger }
-
import kafka.api.TopicMetadata
-
import org.I0Itec.zkclient.ZkClient
import org.apache.samza.util.Clock
import org.apache.samza.util.TopicMetadataStore
import org.junit.Assert._
import org.junit.Before
import org.junit.Test
+import kafka.common.ErrorMapping
+import kafka.api.PartitionMetadata
class TestTopicMetadataCache {
@@ -124,4 +124,14 @@ class TestTopicMetadataCache {
assertTrue(numAssertionSuccess.get())
assertEquals(1, mockStore.numberOfCalls.get())
}
+
+ @Test
+ def testBadErrorCodes {
+ val partitionMetadataBad = new PartitionMetadata(0, None, Seq(), errorCode = ErrorMapping.LeaderNotAvailableCode)
+ val partitionMetadataGood = new PartitionMetadata(0, None, Seq(), errorCode = ErrorMapping.NoError)
+ assertTrue(TopicMetadataCache.hasBadErrorCode(new TopicMetadata("test", List.empty, ErrorMapping.RequestTimedOutCode)))
+ assertTrue(TopicMetadataCache.hasBadErrorCode(new TopicMetadata("test", List(partitionMetadataBad), ErrorMapping.NoError)))
+ assertFalse(TopicMetadataCache.hasBadErrorCode(new TopicMetadata("test", List.empty, ErrorMapping.NoError)))
+ assertFalse(TopicMetadataCache.hasBadErrorCode(new TopicMetadata("test", List(partitionMetadataGood), ErrorMapping.NoError)))
+ }
}
http://git-wip-us.apache.org/repos/asf/samza/blob/1202df41/samza-kafka/src/test/scala/org/apache/samza/utils/TestKafkaUtil.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/test/scala/org/apache/samza/utils/TestKafkaUtil.scala b/samza-kafka/src/test/scala/org/apache/samza/utils/TestKafkaUtil.scala
new file mode 100644
index 0000000..848cfc8
--- /dev/null
+++ b/samza-kafka/src/test/scala/org/apache/samza/utils/TestKafkaUtil.scala
@@ -0,0 +1,33 @@
+/*
+ * 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.samza.utils
+
+import org.junit.Test
+import org.scalatest.{ Matchers => ScalaTestMatchers }
+import org.apache.samza.util.KafkaUtil
+import kafka.common.ErrorMapping
+
+class TestKafkaUtil extends ScalaTestMatchers {
+ @Test
+ def testMaybeThrowException {
+ intercept[Exception] { KafkaUtil.maybeThrowException(ErrorMapping.UnknownTopicOrPartitionCode) }
+ KafkaUtil.maybeThrowException(ErrorMapping.ReplicaNotAvailableCode)
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/samza/blob/1202df41/samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala
----------------------------------------------------------------------
diff --git a/samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala b/samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala
index a8b724b..d66b3bd 100644
--- a/samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala
+++ b/samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala
@@ -22,9 +22,7 @@ package org.apache.samza.test.integration
import java.util.Properties
import java.util.concurrent.CountDownLatch
import java.util.concurrent.TimeUnit
-
import kafka.admin.AdminUtils
-import kafka.common.ErrorMapping
import kafka.consumer.Consumer
import kafka.consumer.ConsumerConfig
import kafka.message.MessageAndMetadata
@@ -35,7 +33,6 @@ import kafka.utils.TestZKUtils
import kafka.utils.Utils
import kafka.utils.ZKStringSerializer
import kafka.zk.EmbeddedZookeeper
-
import org.I0Itec.zkclient.ZkClient
import org.apache.samza.Partition
import org.apache.samza.checkpoint.Checkpoint
@@ -59,14 +56,13 @@ import org.apache.samza.util.ClientUtilTopicMetadataStore
import org.apache.samza.util.TopicMetadataStore
import org.junit.Assert._
import org.junit.{BeforeClass, AfterClass, Test}
-
import scala.collection.JavaConversions._
import scala.collection.mutable.ArrayBuffer
import scala.collection.mutable.HashMap
import scala.collection.mutable.SynchronizedMap
import org.apache.kafka.clients.producer.{ProducerConfig, Producer, ProducerRecord, KafkaProducer}
import java.util
-
+import org.apache.samza.util.KafkaUtil
object TestStatefulTask {
val INPUT_TOPIC = "input"
@@ -145,7 +141,7 @@ object TestStatefulTask {
val topicMetadata = topicMetadataMap(topic)
val errorCode = topicMetadata.errorCode
- ErrorMapping.maybeThrowException(errorCode)
+ KafkaUtil.maybeThrowException(errorCode)
})
done = true
[10/36] samza git commit: SAMZA-597;
support JSON encoding for log4j stream appender
Posted by cr...@apache.org.
SAMZA-597; support JSON encoding for log4j stream appender
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/23c4b391
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/23c4b391
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/23c4b391
Branch: refs/heads/samza-sql
Commit: 23c4b391545e68fa005cedc8db9b01d5270000d6
Parents: 769844e
Author: Chris Riccomini <cr...@apache.org>
Authored: Fri Mar 13 14:40:34 2015 -0700
Committer: Chris Riccomini <cr...@apache.org>
Committed: Fri Mar 13 14:40:34 2015 -0700
----------------------------------------------------------------------
build.gradle | 1 +
.../versioned/jobs/configuration-table.html | 13 ++
.../documentation/versioned/jobs/logging.md | 13 +-
.../apache/samza/config/Log4jSystemConfig.java | 67 +++----
.../samza/logging/log4j/StreamAppender.java | 36 ++--
.../serializers/LoggingEventJsonSerde.java | 192 +++++++++++++++++++
.../LoggingEventJsonSerdeFactory.java | 38 ++++
.../samza/config/TestLog4jSystemConfig.java | 22 ++-
.../samza/logging/log4j/TestStreamAppender.java | 54 +++++-
9 files changed, 367 insertions(+), 69 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/23c4b391/build.gradle
----------------------------------------------------------------------
diff --git a/build.gradle b/build.gradle
index 516c83e..87334fa 100644
--- a/build.gradle
+++ b/build.gradle
@@ -190,6 +190,7 @@ project(':samza-log4j') {
compile "log4j:log4j:$log4jVersion"
compile project(':samza-api')
compile project(":samza-core_$scalaVersion")
+ compile "org.codehaus.jackson:jackson-mapper-asl:$jacksonVersion"
testCompile "junit:junit:$junitVersion"
}
}
http://git-wip-us.apache.org/repos/asf/samza/blob/23c4b391/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 ec12874..e091460 100644
--- a/docs/learn/documentation/versioned/jobs/configuration-table.html
+++ b/docs/learn/documentation/versioned/jobs/configuration-table.html
@@ -388,6 +388,19 @@
</tr>
<tr>
+ <td class="property" id="task-log4j-location-info-enabled">task.log4j.location.info.enabled</td>
+ <td class="default">false</td>
+ <td class="description">
+ Defines whether or not to include log4j's LocationInfo data in Log4j StreamAppender messages. LocationInfo includes
+ information such as the file, class, and line that wrote a log message. This setting is only active if the Log4j
+ stream appender is being used. (See <a href="logging.html#stream-log4j-appender">Stream Log4j Appender</a>)
+ <dl>
+ <dt>Example: <code>task.log4j.location.info.enabled=true</code></dt>
+ </dl>
+ </td>
+ </tr>
+
+ <tr>
<td class="property" id="task-poll-interval-ms">task.poll.interval.ms</td>
<td class="default"></td>
<td class="description">
http://git-wip-us.apache.org/repos/asf/samza/blob/23c4b391/docs/learn/documentation/versioned/jobs/logging.md
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/jobs/logging.md b/docs/learn/documentation/versioned/jobs/logging.md
index af2fd0e..1d13d15 100644
--- a/docs/learn/documentation/versioned/jobs/logging.md
+++ b/docs/learn/documentation/versioned/jobs/logging.md
@@ -97,7 +97,7 @@ And then updating your log4j.xml to include the appender:
#### Stream Log4j Appender
-Samza provides a StreamAppender to publish the logs into a specific system. You can specify the system name using "task.log4j.system". If there is only one system in the config, Samza will use that system for the log publishing. Also we have the [MDC|http://logback.qos.ch/manual/mdc.html] keys "containerName", "jobName" and "jobId", which help identify the source of the log. In order to use this appender, simply add:
+Samza provides a StreamAppender to publish the logs into a specific system. You can specify the system name using "task.log4j.system". If there is only one system in the config, Samza will use that system for the log publishing. Also we have the [MDC](http://logback.qos.ch/manual/mdc.html) keys "containerName", "jobName" and "jobId", which help identify the source of the log. In order to use this appender, simply add:
{% highlight xml %}
<appender name="StreamAppender" class="org.apache.samza.logging.log4j.StreamAppender">
@@ -111,7 +111,16 @@ and add:
{% highlight xml %}
<appender-ref ref="StreamAppender"/>
-{% endhighlight %}.
+{% endhighlight %}
+
+Configuring the StreamAppender will automatically encode messages using logstash's [Log4J JSON format](https://github.com/logstash/log4j-jsonevent-layout). Samza also supports pluggable serialization for those that prefer non-JSON logging events. This can be configured the same way other stream serializers are defined:
+
+{% highlight jproperties %}
+serializers.registry.log4j-string.class=org.apache.samza.logging.log4j.serializers.LoggingEventStringSerdeFactory
+systems.mock.streams.\_\_samza\__jobname_\__jobid_\_logs.samza.msg.serde=log4j-string
+{% endhighlight %}
+
+The StreamAppender will always send messages to a job's log stream keyed by the container name.
### Log Directory
http://git-wip-us.apache.org/repos/asf/samza/blob/23c4b391/samza-log4j/src/main/java/org/apache/samza/config/Log4jSystemConfig.java
----------------------------------------------------------------------
diff --git a/samza-log4j/src/main/java/org/apache/samza/config/Log4jSystemConfig.java b/samza-log4j/src/main/java/org/apache/samza/config/Log4jSystemConfig.java
index 107ddf0..d5e24f2 100644
--- a/samza-log4j/src/main/java/org/apache/samza/config/Log4jSystemConfig.java
+++ b/samza-log4j/src/main/java/org/apache/samza/config/Log4jSystemConfig.java
@@ -20,16 +20,16 @@
package org.apache.samza.config;
import java.util.ArrayList;
+import java.util.List;
import java.util.Map;
-import org.apache.samza.logging.log4j.serializers.LoggingEventStringSerdeFactory;
-
/**
* This class contains the methods for getting properties that are needed by the
* StreamAppender.
*/
public class Log4jSystemConfig {
+ private static final String LOCATION_ENABLED = "task.log4j.location.info.enabled";
private static final String TASK_LOG4J_SYSTEM = "task.log4j.system";
private static final String SYSTEM_PREFIX = "systems.";
private static final String SYSTEM_FACTORY_SUFFIX = ".samza.factory";
@@ -41,30 +41,42 @@ public class Log4jSystemConfig {
}
/**
+ * Defines whether or not to include file location information for Log4J
+ * appender messages. File location information includes the method, line
+ * number, class, etc.
+ *
+ * @return If true, will include file location (method, line number, etc)
+ * information in Log4J appender messages.
+ */
+ public boolean getLocationEnabled() {
+ return "true".equals(config.get(Log4jSystemConfig.LOCATION_ENABLED, "false"));
+ }
+
+ /**
* Get the log4j system name from the config. If it's not defined, try to
* guess the system name if there is only one system is defined.
*
* @return log4j system name
*/
public String getSystemName() {
- String log4jSystem = getValue(TASK_LOG4J_SYSTEM);
+ String log4jSystem = config.get(TASK_LOG4J_SYSTEM, null);
if (log4jSystem == null) {
- ArrayList<String> systemNames = getSystemNames();
+ List<String> systemNames = getSystemNames();
if (systemNames.size() == 1) {
log4jSystem = systemNames.get(0);
} else {
- throw new ConfigException("Missing task.log4j.system configuration, and more than 1 systems were found.");
+ throw new ConfigException("Missing " + TASK_LOG4J_SYSTEM + " configuration, and more than 1 systems were found.");
}
}
return log4jSystem;
}
public String getJobName() {
- return getValue(JobConfig.JOB_NAME());
+ return config.get(JobConfig.JOB_NAME(), null);
}
public String getJobId() {
- return getValue(JobConfig.JOB_ID());
+ return config.get(JobConfig.JOB_ID(), null);
}
public String getSystemFactory(String name) {
@@ -72,47 +84,24 @@ public class Log4jSystemConfig {
return null;
}
String systemFactory = String.format(SystemConfig.SYSTEM_FACTORY(), name);
- return getValue(systemFactory);
+ return config.get(systemFactory, null);
}
/**
- * get the class name according to the serde name. If the serde name is "log4j" and
- * the serde class is not configured, will use the default {@link LoggingEventStringSerdeFactory}
+ * Get the class name according to the serde name.
*
- * @param name serde name
- * @return serde factory name
+ * @param name
+ * serde name
+ * @return serde factory name, or null if there is no factory defined for the
+ * supplied serde name.
*/
public String getSerdeClass(String name) {
- String className = getValue(String.format(SerializerConfig.SERDE(), name));
- if (className == null && name.equals("log4j")) {
- className = LoggingEventStringSerdeFactory.class.getCanonicalName();
- }
- return className;
- }
-
- public String getSystemSerdeName(String name) {
- String systemSerdeNameConfig = String.format(SystemConfig.MSG_SERDE(), name);
- return getValue(systemSerdeNameConfig);
+ return config.get(String.format(SerializerConfig.SERDE(), name), null);
}
public String getStreamSerdeName(String systemName, String streamName) {
String streamSerdeNameConfig = String.format(StreamConfig.MSG_SERDE(), systemName, streamName);
- return getValue(streamSerdeNameConfig);
- }
-
- /**
- * A helper method to get the value from the config. If the config does not
- * contain the key, return null.
- *
- * @param key key name
- * @return value of the key in the config
- */
- protected String getValue(String key) {
- if (config.containsKey(key)) {
- return config.get(key);
- } else {
- return null;
- }
+ return config.get(streamSerdeNameConfig, null);
}
/**
@@ -120,7 +109,7 @@ public class Log4jSystemConfig {
*
* @return A list system names
*/
- protected ArrayList<String> getSystemNames() {
+ protected List<String> getSystemNames() {
Config subConf = config.subset(SYSTEM_PREFIX, true);
ArrayList<String> systemNames = new ArrayList<String>();
for (Map.Entry<String, String> entry : subConf.entrySet()) {
http://git-wip-us.apache.org/repos/asf/samza/blob/23c4b391/samza-log4j/src/main/java/org/apache/samza/logging/log4j/StreamAppender.java
----------------------------------------------------------------------
diff --git a/samza-log4j/src/main/java/org/apache/samza/logging/log4j/StreamAppender.java b/samza-log4j/src/main/java/org/apache/samza/logging/log4j/StreamAppender.java
index 4ef3551..d3f25c0 100644
--- a/samza-log4j/src/main/java/org/apache/samza/logging/log4j/StreamAppender.java
+++ b/samza-log4j/src/main/java/org/apache/samza/logging/log4j/StreamAppender.java
@@ -29,10 +29,10 @@ import org.apache.log4j.spi.LoggingEvent;
import org.apache.samza.SamzaException;
import org.apache.samza.config.Config;
import org.apache.samza.config.Log4jSystemConfig;
+import org.apache.samza.config.SerializerConfig;
import org.apache.samza.config.ShellCommandConfig;
-import org.apache.samza.config.StreamConfig;
-import org.apache.samza.config.SystemConfig;
import org.apache.samza.job.model.JobModel;
+import org.apache.samza.logging.log4j.serializers.LoggingEventJsonSerdeFactory;
import org.apache.samza.metrics.MetricsRegistryMap;
import org.apache.samza.serializers.Serde;
import org.apache.samza.serializers.SerdeFactory;
@@ -83,7 +83,11 @@ public class StreamAppender extends AppenderSkeleton {
@Override
public void activateOptions() {
String containerName = System.getProperty(JAVA_OPTS_CONTAINER_NAME);
- isApplicationMaster = containerName.contains(APPLICATION_MASTER_TAG);
+ if (containerName != null) {
+ isApplicationMaster = containerName.contains(APPLICATION_MASTER_TAG);
+ } else {
+ throw new SamzaException("Got null container name from system property: " + JAVA_OPTS_CONTAINER_NAME + ". This is used as the key for the log appender, so can't proceed.");
+ }
key = containerName; // use the container name as the key for the logs
config = getConfig();
SystemFactory systemFactory = null;
@@ -185,7 +189,7 @@ public class StreamAppender extends AppenderSkeleton {
return config;
}
- private String getStreamName(String jobName, String jobId) {
+ public static String getStreamName(String jobName, String jobId) {
if (jobName == null) {
throw new SamzaException("job name is null. Please specify job.name");
}
@@ -205,24 +209,28 @@ public class StreamAppender extends AppenderSkeleton {
* @param streamName name of the stream
*/
private void setSerde(Log4jSystemConfig log4jSystemConfig, String systemName, String streamName) {
- String serdeClass = null;
+ String serdeClass = LoggingEventJsonSerdeFactory.class.getCanonicalName();;
String serdeName = log4jSystemConfig.getStreamSerdeName(systemName, streamName);
- if (serdeName == null) {
- serdeName = log4jSystemConfig.getSystemSerdeName(systemName);
+ if (serdeName != null) {
+ serdeClass = log4jSystemConfig.getSerdeClass(serdeName);
}
- if (serdeName == null) {
- throw new SamzaException("Missing serde name. Please specify the " + StreamConfig.MSG_SERDE() + " or " + SystemConfig.MSG_SERDE() + " property.");
- }
-
- serdeClass = log4jSystemConfig.getSerdeClass(serdeName);
-
if (serdeClass != null) {
SerdeFactory<LoggingEvent> serdeFactory = Util.<SerdeFactory<LoggingEvent>> getObj(serdeClass);
serde = serdeFactory.getSerde(systemName, config);
} else {
- throw new SamzaException("Can not find serializers class. Please specify serializers.registry.s%.class property");
+ String serdeKey = String.format(SerializerConfig.SERDE(), serdeName);
+ throw new SamzaException("Can not find serializers class for key '" + serdeName + "'. Please specify " + serdeKey + " property");
}
}
+
+ /**
+ * Returns the serde that is being used for the stream appender.
+ *
+ * @return The Serde<LoggingEvent> that the appender is using.
+ */
+ public Serde<LoggingEvent> getSerde() {
+ return serde;
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/samza/blob/23c4b391/samza-log4j/src/main/java/org/apache/samza/logging/log4j/serializers/LoggingEventJsonSerde.java
----------------------------------------------------------------------
diff --git a/samza-log4j/src/main/java/org/apache/samza/logging/log4j/serializers/LoggingEventJsonSerde.java b/samza-log4j/src/main/java/org/apache/samza/logging/log4j/serializers/LoggingEventJsonSerde.java
new file mode 100644
index 0000000..fd1c476
--- /dev/null
+++ b/samza-log4j/src/main/java/org/apache/samza/logging/log4j/serializers/LoggingEventJsonSerde.java
@@ -0,0 +1,192 @@
+/*
+ * 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.samza.logging.log4j.serializers;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.text.Format;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.log4j.spi.LocationInfo;
+import org.apache.log4j.spi.LoggingEvent;
+import org.apache.log4j.spi.ThrowableInformation;
+import org.apache.samza.serializers.JsonSerde;
+import org.apache.samza.serializers.Serde;
+
+/**
+ * A JSON serde that serializes Log4J LoggingEvent objects into JSON using the
+ * standard logstash LoggingEvent format defined <a
+ * href="https://github.com/logstash/log4j-jsonevent-layout">here</a>.
+ */
+public class LoggingEventJsonSerde implements Serde<LoggingEvent> {
+ /**
+ * The JSON format version.
+ */
+ public static final int VERSION = 1;
+
+ /**
+ * The date format to use for the timestamp field.
+ */
+ public static final Format DATE_FORMAT = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
+
+ // Have to wrap rather than extend due to type collisions between
+ // Serde<LoggingEvent> and Serde<Object>.
+ private final JsonSerde jsonSerde;
+
+ /**
+ * Defines whether to include LocationInfo data in the serialized
+ * LoggingEvent. This information includes the file, line, and class that
+ * wrote the log line.
+ */
+ private final boolean includeLocationInfo;
+
+ /**
+ * Constructs the serde without location info.
+ */
+ public LoggingEventJsonSerde() {
+ this(false);
+ }
+
+ /**
+ * Constructs the serde.
+ *
+ * @param includeLocationInfo
+ * Whether to include location info in the logging event or not.
+ */
+ public LoggingEventJsonSerde(boolean includeLocationInfo) {
+ this.includeLocationInfo = includeLocationInfo;
+ this.jsonSerde = new JsonSerde();
+ }
+
+ @Override
+ public byte[] toBytes(LoggingEvent loggingEvent) {
+ Map<String, Object> loggingEventMap = encodeToMap(loggingEvent, includeLocationInfo);
+ return jsonSerde.toBytes(loggingEventMap);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public LoggingEvent fromBytes(byte[] loggingEventMapBytes) {
+ Map<String, Object> loggingEventMap = (Map<String, Object>) jsonSerde.fromBytes(loggingEventMapBytes);
+ return decodeFromMap(loggingEventMap);
+ }
+
+ /**
+ * Encodes a LoggingEvent into a HashMap using the logstash JSON format.
+ *
+ * @param loggingEvent
+ * The LoggingEvent to encode.
+ * @param includeLocationInfo
+ * Whether to include LocationInfo in the map, or not.
+ * @return A Map representing the LoggingEvent, which is suitable to be
+ * serialized by a JSON encoder such as Jackson.
+ */
+ @SuppressWarnings("rawtypes")
+ public static Map<String, Object> encodeToMap(LoggingEvent loggingEvent, boolean includeLocationInfo) {
+ Map<String, Object> logstashEvent = new LoggingEventMap();
+ String threadName = loggingEvent.getThreadName();
+ long timestamp = loggingEvent.getTimeStamp();
+ HashMap<String, Object> exceptionInformation = new HashMap<String, Object>();
+ Map mdc = loggingEvent.getProperties();
+ String ndc = loggingEvent.getNDC();
+
+ logstashEvent.put("@version", VERSION);
+ logstashEvent.put("@timestamp", dateFormat(timestamp));
+ logstashEvent.put("source_host", getHostname());
+ logstashEvent.put("message", loggingEvent.getRenderedMessage());
+
+ if (loggingEvent.getThrowableInformation() != null) {
+ final ThrowableInformation throwableInformation = loggingEvent.getThrowableInformation();
+ if (throwableInformation.getThrowable().getClass().getCanonicalName() != null) {
+ exceptionInformation.put("exception_class", throwableInformation.getThrowable().getClass().getCanonicalName());
+ }
+ if (throwableInformation.getThrowable().getMessage() != null) {
+ exceptionInformation.put("exception_message", throwableInformation.getThrowable().getMessage());
+ }
+ if (throwableInformation.getThrowableStrRep() != null) {
+ StringBuilder stackTrace = new StringBuilder();
+ for (String line : throwableInformation.getThrowableStrRep()) {
+ stackTrace.append(line);
+ stackTrace.append("\n");
+ }
+ exceptionInformation.put("stacktrace", stackTrace);
+ }
+ logstashEvent.put("exception", exceptionInformation);
+ }
+
+ if (includeLocationInfo) {
+ LocationInfo info = loggingEvent.getLocationInformation();
+ logstashEvent.put("file", info.getFileName());
+ logstashEvent.put("line_number", info.getLineNumber());
+ logstashEvent.put("class", info.getClassName());
+ logstashEvent.put("method", info.getMethodName());
+ }
+
+ logstashEvent.put("logger_name", loggingEvent.getLoggerName());
+ logstashEvent.put("mdc", mdc);
+ logstashEvent.put("ndc", ndc);
+ logstashEvent.put("level", loggingEvent.getLevel().toString());
+ logstashEvent.put("thread_name", threadName);
+
+ return logstashEvent;
+ }
+
+ /**
+ * This method is not currently implemented.
+ */
+ public static LoggingEvent decodeFromMap(Map<String, Object> loggingEventMap) {
+ throw new UnsupportedOperationException("Unable to decode LoggingEvents.");
+ }
+
+ public static String dateFormat(long time) {
+ return DATE_FORMAT.format(new Date(time));
+ }
+
+ /**
+ * @return The hostname to use in the hostname field of the encoded
+ * LoggingEvents.
+ */
+ public static String getHostname() {
+ try {
+ return InetAddress.getLocalHost().getHostName();
+ } catch (UnknownHostException e) {
+ return "unknown-host";
+ }
+ }
+
+ /**
+ * A helper class that only puts non-null values into the encoded LoggingEvent
+ * map. This helps to shrink over-the-wire byte payloads for encoded
+ * LoggingEvents.
+ */
+ @SuppressWarnings("serial")
+ public static final class LoggingEventMap extends HashMap<String, Object> {
+ public Object put(String key, Object value) {
+ if (value == null) {
+ return get(key);
+ } else {
+ return super.put(key, value);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/23c4b391/samza-log4j/src/main/java/org/apache/samza/logging/log4j/serializers/LoggingEventJsonSerdeFactory.java
----------------------------------------------------------------------
diff --git a/samza-log4j/src/main/java/org/apache/samza/logging/log4j/serializers/LoggingEventJsonSerdeFactory.java b/samza-log4j/src/main/java/org/apache/samza/logging/log4j/serializers/LoggingEventJsonSerdeFactory.java
new file mode 100644
index 0000000..9fdc8ae
--- /dev/null
+++ b/samza-log4j/src/main/java/org/apache/samza/logging/log4j/serializers/LoggingEventJsonSerdeFactory.java
@@ -0,0 +1,38 @@
+/*
+ * 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.samza.logging.log4j.serializers;
+
+import org.apache.log4j.spi.LoggingEvent;
+import org.apache.samza.config.Config;
+import org.apache.samza.config.Log4jSystemConfig;
+import org.apache.samza.serializers.Serde;
+import org.apache.samza.serializers.SerdeFactory;
+
+/**
+ * A factory that returns a Serde suitable for encoding Log4J LoggingEvents as
+ * JSON-encoded maps.
+ */
+public class LoggingEventJsonSerdeFactory implements SerdeFactory<LoggingEvent> {
+ @Override
+ public Serde<LoggingEvent> getSerde(String name, Config config) {
+ boolean locationInfoEnabled = new Log4jSystemConfig(config).getLocationEnabled();
+ return new LoggingEventJsonSerde(locationInfoEnabled);
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/23c4b391/samza-log4j/src/test/java/org/apache/samza/config/TestLog4jSystemConfig.java
----------------------------------------------------------------------
diff --git a/samza-log4j/src/test/java/org/apache/samza/config/TestLog4jSystemConfig.java b/samza-log4j/src/test/java/org/apache/samza/config/TestLog4jSystemConfig.java
index 16ccb45..6314a3e 100644
--- a/samza-log4j/src/test/java/org/apache/samza/config/TestLog4jSystemConfig.java
+++ b/samza-log4j/src/test/java/org/apache/samza/config/TestLog4jSystemConfig.java
@@ -19,12 +19,12 @@
package org.apache.samza.config;
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
import java.util.HashMap;
import java.util.Map;
-import org.apache.samza.logging.log4j.serializers.LoggingEventStringSerdeFactory;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
@@ -35,14 +35,13 @@ public class TestLog4jSystemConfig {
public ExpectedException exception = ExpectedException.none();
@Test
- public void testGetSystemNamesAndGetValue() {
+ public void testGetSystemNames() {
Map<String, String> map = new HashMap<String, String>();
map.put("systems.system1.samza.factory","1");
map.put("systems.system2.samza.factory","2");
Log4jSystemConfig log4jSystemConfig = new Log4jSystemConfig(new MapConfig(map));
assertEquals(2, log4jSystemConfig.getSystemNames().size());
- assertEquals("1", log4jSystemConfig.getValue("systems.system1.samza.factory"));
}
@Test
@@ -71,20 +70,27 @@ public class TestLog4jSystemConfig {
Map<String, String> map = new HashMap<String, String>();
Log4jSystemConfig log4jSystemConfig = new Log4jSystemConfig(new MapConfig(map));
- // get the default serde
- assertEquals(LoggingEventStringSerdeFactory.class.getCanonicalName(), log4jSystemConfig.getSerdeClass("log4j"));
// get null
assertNull(log4jSystemConfig.getSerdeClass("otherName"));
+
+ // get serde
+ map.put("serializers.registry.log4j.class", "someClass");
+ log4jSystemConfig = new Log4jSystemConfig(new MapConfig(map));
+ assertEquals("someClass", log4jSystemConfig.getSerdeClass("log4j"));
}
@Test
public void testGetSerdeName() {
Map<String, String> map = new HashMap<String, String>();
- map.put("systems.mockSystem.streams.mockStream.samza.msg.serde", "streamSerde");
map.put("systems.mockSystem.samza.msg.serde", "systemSerde");
Log4jSystemConfig log4jSystemConfig = new Log4jSystemConfig(new MapConfig(map));
+ // no stream serde
+ assertNull(log4jSystemConfig.getStreamSerdeName("mockSystem", "mockStream"));
+
+ // stream serde
+ map.put("systems.mockSystem.streams.mockStream.samza.msg.serde", "streamSerde");
+ log4jSystemConfig = new Log4jSystemConfig(new MapConfig(map));
assertEquals("streamSerde", log4jSystemConfig.getStreamSerdeName("mockSystem", "mockStream"));
- assertEquals("systemSerde", log4jSystemConfig.getSystemSerdeName("mockSystem"));
}
}
http://git-wip-us.apache.org/repos/asf/samza/blob/23c4b391/samza-log4j/src/test/java/org/apache/samza/logging/log4j/TestStreamAppender.java
----------------------------------------------------------------------
diff --git a/samza-log4j/src/test/java/org/apache/samza/logging/log4j/TestStreamAppender.java b/samza-log4j/src/test/java/org/apache/samza/logging/log4j/TestStreamAppender.java
index 3e4ddc9..3e81240 100644
--- a/samza-log4j/src/test/java/org/apache/samza/logging/log4j/TestStreamAppender.java
+++ b/samza-log4j/src/test/java/org/apache/samza/logging/log4j/TestStreamAppender.java
@@ -28,6 +28,8 @@ import org.apache.log4j.Logger;
import org.apache.log4j.PatternLayout;
import org.apache.samza.config.Config;
import org.apache.samza.config.MapConfig;
+import org.apache.samza.logging.log4j.serializers.LoggingEventJsonSerde;
+import org.apache.samza.logging.log4j.serializers.LoggingEventStringSerde;
import org.apache.samza.logging.log4j.serializers.LoggingEventStringSerdeFactory;
import org.junit.Test;
@@ -36,6 +38,37 @@ public class TestStreamAppender {
static Logger log = Logger.getLogger(TestStreamAppender.class);
@Test
+ public void testDefaultSerde() {
+ System.setProperty("samza.container.name", "samza-container-1");
+ MockSystemProducerAppender systemProducerAppender = new MockSystemProducerAppender();
+ PatternLayout layout = new PatternLayout();
+ layout.setConversionPattern("%m");
+ systemProducerAppender.setLayout(layout);
+ systemProducerAppender.activateOptions();
+ assertNotNull(systemProducerAppender.getSerde());
+ assertEquals(LoggingEventJsonSerde.class, systemProducerAppender.getSerde().getClass());
+ }
+
+ @Test
+ public void testNonDefaultSerde() {
+ System.setProperty("samza.container.name", "samza-container-1");
+ String streamName = StreamAppender.getStreamName("log4jTest", "1");
+ Map<String, String> map = new HashMap<String, String>();
+ map.put("job.name", "log4jTest");
+ map.put("job.id", "1");
+ map.put("serializers.registry.log4j-string.class", LoggingEventStringSerdeFactory.class.getCanonicalName());
+ map.put("systems.mock.samza.factory", MockSystemFactory.class.getCanonicalName());
+ map.put("systems.mock.streams." + streamName + ".samza.msg.serde", "log4j-string");
+ MockSystemProducerAppender systemProducerAppender = new MockSystemProducerAppender(new MapConfig(map));
+ PatternLayout layout = new PatternLayout();
+ layout.setConversionPattern("%m");
+ systemProducerAppender.setLayout(layout);
+ systemProducerAppender.activateOptions();
+ assertNotNull(systemProducerAppender.getSerde());
+ assertEquals(LoggingEventStringSerde.class, systemProducerAppender.getSerde().getClass());
+ }
+
+ @Test
public void testSystemProducerAppender() {
System.setProperty("samza.container.name", "samza-container-1");
@@ -51,8 +84,8 @@ public class TestStreamAppender {
systemProducerAppender.flushSystemProducer();
assertEquals(2, MockSystemProducer.messagesReceived.size());
- assertEquals("testing", new String((byte[])MockSystemProducer.messagesReceived.get(0)));
- assertEquals("testing2", new String((byte[])MockSystemProducer.messagesReceived.get(1)));
+ assertTrue(new String((byte[]) MockSystemProducer.messagesReceived.get(0)).contains("\"message\":\"testing\""));
+ assertTrue(new String((byte[]) MockSystemProducer.messagesReceived.get(1)).contains("\"message\":\"testing2\""));
}
/**
@@ -61,13 +94,22 @@ public class TestStreamAppender {
* stays is difficult to test.
*/
class MockSystemProducerAppender extends StreamAppender {
- @Override
- protected Config getConfig() {
+ private final Config config;
+
+ public MockSystemProducerAppender() {
Map<String, String> map = new HashMap<String, String>();
map.put("job.name", "log4jTest");
map.put("systems.mock.samza.factory", MockSystemFactory.class.getCanonicalName());
- map.put("systems.mock.samza.msg.serde", "log4j");
- return new MapConfig(map);
+ config = new MapConfig(map);
+ }
+
+ public MockSystemProducerAppender(Config config) {
+ this.config = config;
+ }
+
+ @Override
+ protected Config getConfig() {
+ return config;
}
}
}
\ No newline at end of file
[06/36] samza git commit: SAMZA-590;
kafka broker proxy should relinquish ownership for all partitions
when its consumer fails
Posted by cr...@apache.org.
SAMZA-590; kafka broker proxy should relinquish ownership for all partitions when its consumer fails
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/c725b3ce
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/c725b3ce
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/c725b3ce
Branch: refs/heads/samza-sql
Commit: c725b3ceb6d2e116f5da257ba45fe70fa877747a
Parents: b82d458
Author: Chris Riccomini <cr...@apache.org>
Authored: Wed Mar 11 09:28:58 2015 -0700
Committer: Chris Riccomini <cr...@apache.org>
Committed: Wed Mar 11 09:28:58 2015 -0700
----------------------------------------------------------------------
.../apache/samza/system/kafka/BrokerProxy.scala | 28 +++++++---
.../samza/system/kafka/TestBrokerProxy.scala | 59 ++++++++++++++++++++
2 files changed, 80 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/c725b3ce/samza-kafka/src/main/scala/org/apache/samza/system/kafka/BrokerProxy.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/scala/org/apache/samza/system/kafka/BrokerProxy.scala b/samza-kafka/src/main/scala/org/apache/samza/system/kafka/BrokerProxy.scala
index f768263..c6e231a 100644
--- a/samza-kafka/src/main/scala/org/apache/samza/system/kafka/BrokerProxy.scala
+++ b/samza-kafka/src/main/scala/org/apache/samza/system/kafka/BrokerProxy.scala
@@ -153,8 +153,9 @@ class BrokerProxy(
},
(exception, loop) => {
- warn("Restarting consumer due to %s. Turn on debugging to get a full stack trace." format exception)
+ warn("Restarting consumer due to %s. Releasing ownership of all partitions, and restarting consumer. Turn on debugging to get a full stack trace." format exception)
debug("Exception detail:", exception)
+ abdicateAll
reconnect = true
})
} catch {
@@ -182,7 +183,6 @@ class BrokerProxy(
nonErrorResponses.foreach { nonError => moveMessagesToTheirQueue(nonError.getKey, nonError.getValue) }
} else {
-
refreshLatencyMetrics
debug("No topic/partitions need to be fetched for %s:%s right now. Sleeping %sms." format (host, port, sleepMSWhileNoTopicPartitions))
@@ -193,13 +193,27 @@ class BrokerProxy(
}
}
- def handleErrors(errorResponses: mutable.Set[Entry[TopicAndPartition, FetchResponsePartitionData]], response:FetchResponse) = {
+ /**
+ * Releases ownership for a single TopicAndPartition. The
+ * KafkaSystemConsumer will try and find a new broker for the
+ * TopicAndPartition.
+ */
+ def abdicate(tp: TopicAndPartition) = removeTopicPartition(tp) match {
// Need to be mindful of a tp that was removed by another thread
- def abdicate(tp:TopicAndPartition) = removeTopicPartition(tp) match {
- case Some(offset) => messageSink.abdicate(tp, offset)
- case None => warn("Tried to abdicate for topic partition not in map. Removed in interim?")
- }
+ case Some(offset) => messageSink.abdicate(tp, offset)
+ case None => warn("Tried to abdicate for topic partition not in map. Removed in interim?")
+ }
+ /**
+ * Releases all TopicAndPartition ownership for this BrokerProxy thread. The
+ * KafkaSystemConsumer will try and find a new broker for the
+ * TopicAndPartition.
+ */
+ def abdicateAll {
+ nextOffsets.keySet.foreach(abdicate(_))
+ }
+
+ def handleErrors(errorResponses: mutable.Set[Entry[TopicAndPartition, FetchResponsePartitionData]], response:FetchResponse) = {
// FetchResponse should really return Option and a list of the errors so we don't have to find them ourselves
case class Error(tp: TopicAndPartition, code: Short, exception: Throwable)
http://git-wip-us.apache.org/repos/asf/samza/blob/c725b3ce/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestBrokerProxy.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestBrokerProxy.scala b/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestBrokerProxy.scala
index d559d8b..e285dec 100644
--- a/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestBrokerProxy.scala
+++ b/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestBrokerProxy.scala
@@ -21,6 +21,7 @@
package org.apache.samza.system.kafka
import java.nio.ByteBuffer
+import java.nio.channels.ClosedChannelException
import java.util.concurrent.CountDownLatch
import kafka.api._
@@ -299,4 +300,62 @@ class TestBrokerProxy extends Logging {
fail(failString)
}
}
+
+ /**
+ * Test that makes sure that BrokerProxy abdicates all TopicAndPartitions
+ * that it owns when a consumer failure occurs.
+ */
+ @Test def brokerProxyAbdicatesOnConnectionFailure():Unit = {
+ val countdownLatch = new CountDownLatch(1)
+ var abdicated: Option[TopicAndPartition] = None
+ @volatile var refreshDroppedCount = 0
+ val mockMessageSink = new MessageSink {
+ override def setIsAtHighWatermark(tp: TopicAndPartition, isAtHighWatermark: Boolean) {
+ }
+ override def addMessage(tp: TopicAndPartition, msg: MessageAndOffset, highWatermark: Long) {
+ }
+ override def abdicate(tp: TopicAndPartition, nextOffset: Long) {
+ abdicated = Some(tp)
+ countdownLatch.countDown
+ }
+ override def refreshDropped() {
+ refreshDroppedCount += 1
+ }
+ override def needsMoreMessages(tp: TopicAndPartition): Boolean = {
+ true
+ }
+ }
+
+ val doNothingMetrics = new KafkaSystemConsumerMetrics()
+ val tp = new TopicAndPartition("topic", 42)
+ val mockOffsetGetter = mock(classOf[GetOffset])
+ val mockSimpleConsumer = mock(classOf[DefaultFetchSimpleConsumer])
+
+ when(mockOffsetGetter.isValidOffset(any(classOf[DefaultFetchSimpleConsumer]), Matchers.eq(tp), Matchers.eq("0"))).thenReturn(true)
+ when(mockOffsetGetter.getResetOffset(any(classOf[DefaultFetchSimpleConsumer]), Matchers.eq(tp))).thenReturn(1492l)
+ when(mockSimpleConsumer.defaultFetch(any())).thenThrow(new SamzaException("Pretend this is a ClosedChannelException. Can't use ClosedChannelException because it's checked, and Mockito doesn't like that."))
+
+ val bp = new BrokerProxy("host", 567, "system", "clientID", doNothingMetrics, mockMessageSink, Int.MaxValue, 1024000, new StreamFetchSizes(256 * 1024), 524288, 1000, mockOffsetGetter) {
+ override def createSimpleConsumer() = {
+ mockSimpleConsumer
+ }
+ }
+
+ val waitForRefresh = () => {
+ val currentRefreshDroppedCount = refreshDroppedCount
+ while (refreshDroppedCount == currentRefreshDroppedCount) {
+ Thread.sleep(100)
+ }
+ }
+
+ bp.addTopicPartition(tp, Option("0"))
+ bp.start
+ // BP should refresh on startup.
+ waitForRefresh()
+ countdownLatch.await()
+ // BP should continue refreshing after it's abdicated all TopicAndPartitions.
+ waitForRefresh()
+ bp.stop
+ assertEquals(tp, abdicated.getOrElse(null))
+ }
}
[22/36] samza git commit: SAMZA-458;
closing kafka system producers flushes all buffered messages
Posted by cr...@apache.org.
SAMZA-458; closing kafka system producers flushes all buffered messages
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/28e2dbb8
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/28e2dbb8
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/28e2dbb8
Branch: refs/heads/samza-sql
Commit: 28e2dbb895d1a9bb4d5d2662966902c7d85546be
Parents: 8b52e8a
Author: Yan Fang <ya...@gmail.com>
Authored: Mon Mar 23 12:41:47 2015 -0700
Committer: Chris Riccomini <cr...@apache.org>
Committed: Mon Mar 23 12:41:47 2015 -0700
----------------------------------------------------------------------
.../system/kafka/KafkaSystemProducer.scala | 10 ++++--
.../system/kafka/TestKafkaSystemProducer.scala | 33 ++++++++++++++++++--
2 files changed, 37 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/28e2dbb8/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemProducer.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemProducer.scala b/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemProducer.scala
index 83668dd..19bc37d 100644
--- a/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemProducer.scala
+++ b/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemProducer.scala
@@ -33,6 +33,7 @@ import org.apache.kafka.common.errors.RetriableException
import org.apache.kafka.common.PartitionInfo
import java.util
import java.util.concurrent.Future
+import scala.collection.JavaConversions._
class KafkaSystemProducer(systemName: String,
@@ -51,6 +52,7 @@ class KafkaSystemProducer(systemName: String,
def stop() {
if (producer != null) {
+ latestFuture.keys.foreach(flush(_))
producer.close
}
}
@@ -136,11 +138,13 @@ class KafkaSystemProducer(systemName: String,
}
if (sendFailed.get()) {
logger.error("Unable to send message from %s to system %s" format(source, systemName))
- //Close producer
- stop()
+ //Close producer.
+ if (producer != null) {
+ producer.close
+ }
producer = null
metrics.flushFailed.inc
- throw new SamzaException("Unable to send message from %s to system %s" format(source, systemName))
+ throw new SamzaException("Unable to send message from %s to system %s." format(source, systemName), exceptionThrown.get)
} else {
trace("Flushed %s." format (source))
}
http://git-wip-us.apache.org/repos/asf/samza/blob/28e2dbb8/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemProducer.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemProducer.scala b/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemProducer.scala
index ca10ea5..ef5a55b 100644
--- a/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemProducer.scala
+++ b/samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemProducer.scala
@@ -133,8 +133,8 @@ class TestKafkaSystemProducer {
assertEquals(1, mockProducer.getMsgsSent)
mockProducer.startDelayedSendThread(2000)
val thrown = intercept[SamzaException] {
- systemProducer.flush("test")
- }
+ systemProducer.flush("test")
+ }
assertTrue(thrown.isInstanceOf[SamzaException])
assertEquals(2, mockProducer.getMsgsSent)
systemProducer.stop()
@@ -162,9 +162,36 @@ class TestKafkaSystemProducer {
producer.send("test", msg4)
}
assertTrue(thrown.isInstanceOf[SamzaException])
- assertTrue(thrown.getMessage.contains("RecordTooLargeException"))
+ assertTrue(thrown.getCause.isInstanceOf[RecordTooLargeException])
assertEquals(true, producer.sendFailed.get())
assertEquals(3, mockProducer.getMsgsSent)
producer.stop()
}
+
+ @Test
+ def testKafkaProducerFlushMsgsWhenStop {
+ val msg1 = new OutgoingMessageEnvelope(new SystemStream("test", "test"), "a".getBytes)
+ val msg2 = new OutgoingMessageEnvelope(new SystemStream("test", "test"), "b".getBytes)
+ val msg3 = new OutgoingMessageEnvelope(new SystemStream("test", "test"), "c".getBytes)
+ val msg4 = new OutgoingMessageEnvelope(new SystemStream("test2", "test"), "d".getBytes)
+
+ val mockProducer = new MockKafkaProducer(1, "test", 1)
+ val systemProducer = new KafkaSystemProducer(systemName = "test",
+ getProducer = () => mockProducer,
+ metrics = new KafkaSystemProducerMetrics)
+ systemProducer.register("test")
+ systemProducer.register("test2")
+ systemProducer.start()
+ systemProducer.send("test", msg1)
+
+ mockProducer.setShouldBuffer(true)
+ systemProducer.send("test", msg2)
+ systemProducer.send("test", msg3)
+ systemProducer.send("test2", msg4)
+ assertEquals(1, mockProducer.getMsgsSent)
+ mockProducer.startDelayedSendThread(2000)
+ assertEquals(1, mockProducer.getMsgsSent)
+ systemProducer.stop()
+ assertEquals(4, mockProducer.getMsgsSent)
+ }
}
[03/36] samza git commit: SAMZA-584;
fix race condition in kafka system consumer
Posted by cr...@apache.org.
SAMZA-584; fix race condition in kafka system consumer
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/fcb5cea3
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/fcb5cea3
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/fcb5cea3
Branch: refs/heads/samza-sql
Commit: fcb5cea3fc4767ba6239a5cdf05a5a06a7f92e62
Parents: 7a21bef
Author: Chris Riccomini <cr...@apache.org>
Authored: Tue Mar 3 10:20:15 2015 -0800
Committer: Chris Riccomini <cr...@apache.org>
Committed: Tue Mar 3 10:20:15 2015 -0800
----------------------------------------------------------------------
.../system/kafka/KafkaSystemConsumer.scala | 27 ++++++++++----------
1 file changed, 13 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/fcb5cea3/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemConsumer.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemConsumer.scala b/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemConsumer.scala
index 38117e2..de00320 100644
--- a/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemConsumer.scala
+++ b/samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemConsumer.scala
@@ -142,7 +142,6 @@ private[kafka] class KafkaSystemConsumer(
// This avoids trying to re-add the same topic partition repeatedly
def refresh(tp: List[TopicAndPartition]) = {
val head :: rest = tpToRefresh
- val nextOffset = topicPartitionsAndOffsets.get(head).get
// refreshBrokers can be called from abdicate and refreshDropped,
// both of which are triggered from BrokerProxy threads. To prevent
// accidentally creating multiple objects for the same broker, or
@@ -151,18 +150,18 @@ private[kafka] class KafkaSystemConsumer(
this.synchronized {
// Check if we still need this TopicAndPartition inside the
// critical section. If we don't, then skip it.
- if (topicPartitionsAndOffsets.contains(head)) {
- getHostPort(topicMetadata(head.topic), head.partition) match {
- case Some((host, port)) =>
- val brokerProxy = brokerProxies.getOrElseUpdate((host, port), createBrokerProxy(host, port))
- brokerProxy.addTopicPartition(head, Option(nextOffset))
- brokerProxy.start
- debug("Claimed topic-partition (%s) for (%s)".format(head, brokerProxy))
- topicPartitionsAndOffsets -= head
- case None => info("No metadata available for: %s. Will try to refresh and add to a consumer thread later." format head)
- }
- } else {
- debug("Ignoring refresh for %s because we already added it from another thread." format head)
+ topicPartitionsAndOffsets.get(head) match {
+ case Some(nextOffset) =>
+ getHostPort(topicMetadata(head.topic), head.partition) match {
+ case Some((host, port)) =>
+ val brokerProxy = brokerProxies.getOrElseUpdate((host, port), createBrokerProxy(host, port))
+ brokerProxy.addTopicPartition(head, Option(nextOffset))
+ brokerProxy.start
+ debug("Claimed topic-partition (%s) for (%s)".format(head, brokerProxy))
+ topicPartitionsAndOffsets -= head
+ case None => info("No metadata available for: %s. Will try to refresh and add to a consumer thread later." format head)
+ }
+ case _ => debug("Ignoring refresh for %s because we already added it from another thread." format head)
}
}
rest
@@ -182,7 +181,7 @@ private[kafka] class KafkaSystemConsumer(
}
val sink = new MessageSink {
- var lastDroppedRefresh = 0L
+ var lastDroppedRefresh = clock()
def refreshDropped() {
if (topicPartitionsAndOffsets.size > 0 && clock() - lastDroppedRefresh > 10000) {
[25/36] samza git commit: SAMZA-588;
expose SamzaContainerContext through TaskContext
Posted by cr...@apache.org.
SAMZA-588; expose SamzaContainerContext through TaskContext
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/2c9020b7
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/2c9020b7
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/2c9020b7
Branch: refs/heads/samza-sql
Commit: 2c9020b7f20d609113868c224fb8138c36521e0e
Parents: 91289f4
Author: Benjamin Fradet <be...@gmail.com>
Authored: Tue Mar 24 12:47:00 2015 -0700
Committer: Chris Riccomini <cr...@apache.org>
Committed: Tue Mar 24 12:47:00 2015 -0700
----------------------------------------------------------------------
.../java/org/apache/samza/task/TaskContext.java | 3 +++
.../apache/samza/container/SamzaContainer.scala | 3 ++-
.../apache/samza/container/TaskInstance.scala | 2 ++
.../samza/container/TestSamzaContainer.scala | 5 +++-
.../samza/container/TestTaskInstance.scala | 25 ++++++++++++++------
5 files changed, 29 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/2c9020b7/samza-api/src/main/java/org/apache/samza/task/TaskContext.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/task/TaskContext.java b/samza-api/src/main/java/org/apache/samza/task/TaskContext.java
index 929409e..5b337a6 100644
--- a/samza-api/src/main/java/org/apache/samza/task/TaskContext.java
+++ b/samza-api/src/main/java/org/apache/samza/task/TaskContext.java
@@ -19,6 +19,7 @@
package org.apache.samza.task;
+import org.apache.samza.container.SamzaContainerContext;
import org.apache.samza.container.TaskName;
import org.apache.samza.metrics.MetricsRegistry;
import org.apache.samza.system.SystemStreamPartition;
@@ -38,6 +39,8 @@ public interface TaskContext {
TaskName getTaskName();
+ SamzaContainerContext getSamzaContainerContext();
+
/**
* Set the starting offset for the given {@link org.apache.samza.system.SystemStreamPartition}. Offsets
* can only be set for a {@link org.apache.samza.system.SystemStreamPartition} assigned to this task
http://git-wip-us.apache.org/repos/asf/samza/blob/2c9020b7/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
index 9fc3b55..5416dd6 100644
--- a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
+++ b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
@@ -414,7 +414,7 @@ object SamzaContainer extends Logging {
// Increment by 1 because partition starts from 0, but we need the absolute count,
// this value is used for change log topic creation.
val maxChangeLogStreamPartitions = containerModel.getTasks.values
- .max(Ordering.by{task:TaskModel => task.getChangelogPartition.getPartitionId})
+ .max(Ordering.by { task:TaskModel => task.getChangelogPartition.getPartitionId })
.getChangelogPartition.getPartitionId + 1
val taskInstances: Map[TaskName, TaskInstance] = containerModel.getTasks.values.map(taskModel => {
@@ -494,6 +494,7 @@ object SamzaContainer extends Logging {
metrics = taskInstanceMetrics,
consumerMultiplexer = consumerMultiplexer,
collector = collector,
+ containerContext = containerContext,
offsetManager = offsetManager,
storageManager = storageManager,
reporters = reporters,
http://git-wip-us.apache.org/repos/asf/samza/blob/2c9020b7/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala b/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala
index be0b55a..c5a5ea5 100644
--- a/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala
+++ b/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala
@@ -46,6 +46,7 @@ class TaskInstance(
metrics: TaskInstanceMetrics,
consumerMultiplexer: SystemConsumers,
collector: TaskInstanceCollector,
+ containerContext: SamzaContainerContext,
offsetManager: OffsetManager = new OffsetManager,
storageManager: TaskStorageManager = null,
reporters: Map[String, MetricsReporter] = Map(),
@@ -65,6 +66,7 @@ class TaskInstance(
null
}
def getTaskName = taskName
+ def getSamzaContainerContext = containerContext
override def setStartingOffset(ssp: SystemStreamPartition, offset: String): Unit = {
offsetManager.startingOffsets += (ssp -> offset)
http://git-wip-us.apache.org/repos/asf/samza/blob/2c9020b7/samza-core/src/test/scala/org/apache/samza/container/TestSamzaContainer.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/container/TestSamzaContainer.scala b/samza-core/src/test/scala/org/apache/samza/container/TestSamzaContainer.scala
index 81742bc..cab31ca 100644
--- a/samza-core/src/test/scala/org/apache/samza/container/TestSamzaContainer.scala
+++ b/samza-core/src/test/scala/org/apache/samza/container/TestSamzaContainer.scala
@@ -121,13 +121,16 @@ class TestSamzaContainer extends AssertionsForJUnit {
Map[String, SystemProducer](),
new SerdeManager)
val collector = new TaskInstanceCollector(producerMultiplexer)
+ val containerContext = new SamzaContainerContext(0, config, Set[TaskName](taskName))
val taskInstance: TaskInstance = new TaskInstance(
task,
taskName,
config,
new TaskInstanceMetrics,
consumerMultiplexer,
- collector)
+ collector,
+ containerContext
+ )
val runLoop = new RunLoop(
taskInstances = Map(taskName -> taskInstance),
consumerMultiplexer = consumerMultiplexer,
http://git-wip-us.apache.org/repos/asf/samza/blob/2c9020b7/samza-core/src/test/scala/org/apache/samza/container/TestTaskInstance.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/container/TestTaskInstance.scala b/samza-core/src/test/scala/org/apache/samza/container/TestTaskInstance.scala
index 54b4df8..7caad28 100644
--- a/samza-core/src/test/scala/org/apache/samza/container/TestTaskInstance.scala
+++ b/samza-core/src/test/scala/org/apache/samza/container/TestTaskInstance.scala
@@ -69,6 +69,7 @@ class TestTaskInstance {
val offsetManager = OffsetManager(Map(systemStream -> testSystemStreamMetadata), config)
val taskName = new TaskName("taskName")
val collector = new TaskInstanceCollector(producerMultiplexer)
+ val containerContext = new SamzaContainerContext(0, config, Set[TaskName](taskName))
val taskInstance: TaskInstance = new TaskInstance(
task,
taskName,
@@ -76,6 +77,7 @@ class TestTaskInstance {
new TaskInstanceMetrics,
consumerMultiplexer,
collector,
+ containerContext,
offsetManager)
// Pretend we got a message with offset 2 and next offset 3.
val coordinator = new ReadableCoordinator(taskName)
@@ -159,6 +161,7 @@ class TestTaskInstance {
val offsetManager = OffsetManager(Map(systemStream -> testSystemStreamMetadata), config)
val taskName = new TaskName("taskName")
val collector = new TaskInstanceCollector(producerMultiplexer)
+ val containerContext = new SamzaContainerContext(0, config, Set[TaskName](taskName))
val registry = new MetricsRegistryMap
val taskMetrics = new TaskInstanceMetrics(registry = registry)
@@ -169,6 +172,7 @@ class TestTaskInstance {
taskMetrics,
consumerMultiplexer,
collector,
+ containerContext,
offsetManager,
exceptionHandler = TaskInstanceExceptionHandler(taskMetrics, config))
@@ -211,6 +215,7 @@ class TestTaskInstance {
val offsetManager = OffsetManager(Map(systemStream -> testSystemStreamMetadata), config)
val taskName = new TaskName("taskName")
val collector = new TaskInstanceCollector(producerMultiplexer)
+ val containerContext = new SamzaContainerContext(0, config, Set[TaskName](taskName))
val registry = new MetricsRegistryMap
val taskMetrics = new TaskInstanceMetrics(registry = registry)
@@ -221,6 +226,7 @@ class TestTaskInstance {
taskMetrics,
consumerMultiplexer,
collector,
+ containerContext,
offsetManager,
exceptionHandler = TaskInstanceExceptionHandler(taskMetrics, config))
@@ -261,23 +267,28 @@ class TestTaskInstance {
override def process(envelope: IncomingMessageEnvelope, collector: MessageCollector, coordinator: TaskCoordinator): Unit = {}
}
+ val config = new MapConfig()
val chooser = new RoundRobinChooser()
val consumers = new SystemConsumers(chooser, consumers = Map.empty)
val producers = new SystemProducers(Map.empty, new SerdeManager())
+ val metrics = new TaskInstanceMetrics()
+ val taskName = new TaskName("Offset Reset Task 0")
val collector = new TaskInstanceCollector(producers)
+ val containerContext = new SamzaContainerContext(0, config, Set[TaskName](taskName))
val offsetManager = new OffsetManager()
offsetManager.startingOffsets ++= Map(partition0 -> "0", partition1 -> "0")
val taskInstance = new TaskInstance(
- task = task,
- taskName = new TaskName("Offset Reset Task 0"),
- config = new MapConfig(),
- metrics = new TaskInstanceMetrics(),
- consumerMultiplexer = consumers,
- collector = collector,
- offsetManager = offsetManager,
+ task,
+ taskName,
+ config,
+ metrics,
+ consumers,
+ collector,
+ containerContext,
+ offsetManager,
systemStreamPartitions = Set(partition0, partition1) )
taskInstance.initTask
[33/36] samza git commit: SAMZA-638: Add Stream Name to
SystemConsumersException Message
Posted by cr...@apache.org.
SAMZA-638: Add Stream Name to SystemConsumersException Message
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/457a5245
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/457a5245
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/457a5245
Branch: refs/heads/samza-sql
Commit: 457a5245100749113b57585e559469530d25caa9
Parents: 76e151c
Author: Navina Ramesh <na...@gmail.com>
Authored: Mon Apr 6 16:17:40 2015 -0700
Committer: Jakob Homan <jg...@gmail.com>
Committed: Mon Apr 6 16:17:40 2015 -0700
----------------------------------------------------------------------
.../main/scala/org/apache/samza/system/SystemConsumers.scala | 7 +++++--
1 file changed, 5 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/457a5245/samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala b/samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala
index 125d376..76fa4ad 100644
--- a/samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala
+++ b/samza-core/src/main/scala/org/apache/samza/system/SystemConsumers.scala
@@ -290,9 +290,12 @@ class SystemConsumers(
Some(serdeManager.fromBytes(rawEnvelope))
} catch {
case e: Exception if !dropDeserializationError =>
- throw new SystemConsumersException("Cannot deserialize an incoming message.", e)
+ throw new SystemConsumersException(
+ "Cannot deserialize an incoming message for %s"
+ .format(systemStreamPartition.getSystemStream.toString), e)
case ex: Exception =>
- debug("Cannot deserialize an incoming message. Dropping the error message.", ex)
+ debug("Cannot deserialize an incoming message for %s. Dropping the error message."
+ .format(systemStreamPartition.getSystemStream.toString), ex)
metrics.deserializationError.inc
None
}
[05/36] samza git commit: SAMZA-506;
gracefully shutdown container on SIGTERM
Posted by cr...@apache.org.
SAMZA-506; gracefully shutdown container on SIGTERM
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/b82d4587
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/b82d4587
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/b82d4587
Branch: refs/heads/samza-sql
Commit: b82d458718e43a4a7a580f5a8e65ee7969b0e0f0
Parents: ff40b12
Author: Ewen Cheslack-Postava <me...@ewencp.org>
Authored: Wed Mar 11 09:06:53 2015 -0700
Committer: Chris Riccomini <cr...@apache.org>
Committed: Wed Mar 11 09:06:53 2015 -0700
----------------------------------------------------------------------
.../org/apache/samza/container/RunLoop.scala | 31 +++++++++++++++---
.../apache/samza/container/TestRunLoop.scala | 34 ++++++++++++++++++++
2 files changed, 60 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/b82d4587/samza-core/src/main/scala/org/apache/samza/container/RunLoop.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/container/RunLoop.scala b/samza-core/src/main/scala/org/apache/samza/container/RunLoop.scala
index 499f5c6..4098235 100644
--- a/samza-core/src/main/scala/org/apache/samza/container/RunLoop.scala
+++ b/samza-core/src/main/scala/org/apache/samza/container/RunLoop.scala
@@ -45,7 +45,7 @@ class RunLoop(
private var lastCommitMs = 0L
private var taskShutdownRequests: Set[TaskName] = Set()
private var taskCommitRequests: Set[TaskName] = Set()
- private var shutdownNow = false
+ @volatile private var shutdownNow = false
// Messages come from the chooser with no connection to the TaskInstance they're bound for.
// Keep a mapping of SystemStreamPartition to TaskInstance to efficiently route them.
@@ -56,15 +56,36 @@ class RunLoop(
taskInstances.values.map { getSystemStreamPartitionToTaskInstance }.flatten.toMap
}
+ val shutdownHook = new Thread() {
+ override def run() = {
+ info("Triggering shutdown in response to shutdown hook")
+ shutdownNow = true
+ }
+ }
+
+ protected def addShutdownHook() {
+ Runtime.getRuntime().addShutdownHook(shutdownHook)
+ }
+
+ protected def removeShutdownHook() {
+ Runtime.getRuntime().removeShutdownHook(shutdownHook)
+ }
+
/**
* Starts the run loop. Blocks until either the tasks request shutdown, or an
* unhandled exception is thrown.
*/
def run {
- while (!shutdownNow) {
- process
- window
- commit
+ try {
+ addShutdownHook()
+
+ while (!shutdownNow) {
+ process
+ window
+ commit
+ }
+ } finally {
+ removeShutdownHook()
}
}
http://git-wip-us.apache.org/repos/asf/samza/blob/b82d4587/samza-core/src/test/scala/org/apache/samza/container/TestRunLoop.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/container/TestRunLoop.scala b/samza-core/src/test/scala/org/apache/samza/container/TestRunLoop.scala
index ea48853..fb1ebdd 100644
--- a/samza-core/src/test/scala/org/apache/samza/container/TestRunLoop.scala
+++ b/samza-core/src/test/scala/org/apache/samza/container/TestRunLoop.scala
@@ -211,4 +211,38 @@ class TestRunLoop extends AssertionsForJUnit with MockitoSugar with ShouldMatche
testMetrics.processMs.getSnapshot.getSize should equal(2)
testMetrics.commitMs.getSnapshot.getSize should equal(2)
}
+
+ @Test
+ def testShutdownHook: Unit = {
+ // The shutdown hook can't be directly tested so we verify that a) both add and remove
+ // are called and b) invoking the shutdown hook actually kills the run loop.
+ val consumers = mock[SystemConsumers]
+ when(consumers.choose).thenReturn(envelope0)
+ val testMetrics = new SamzaContainerMetrics
+ var addCalled = false
+ var removeCalled = false
+ val runLoop = new RunLoop(
+ taskInstances = getMockTaskInstances,
+ consumerMultiplexer = consumers,
+ metrics = testMetrics) {
+ override def addShutdownHook() {
+ addCalled = true
+ }
+ override def removeShutdownHook() {
+ removeCalled = true
+ }
+ }
+
+ val runThread = new Thread(runLoop)
+ runThread.start()
+
+ runLoop.shutdownHook.start()
+ runLoop.shutdownHook.join(1000)
+ runThread.join(1000)
+
+ assert(addCalled)
+ assert(removeCalled)
+ assert(!runLoop.shutdownHook.isAlive)
+ assert(!runThread.isAlive)
+ }
}
[16/36] samza git commit: SAMZA-568;
allow tasks to override offsets in init method
Posted by cr...@apache.org.
SAMZA-568; allow tasks to override offsets in init method
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/537262e2
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/537262e2
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/537262e2
Branch: refs/heads/samza-sql
Commit: 537262e284d6c126ea434c1af74e66e92cf08e5e
Parents: cad2c82
Author: Ben Kirwin <be...@kirw.in>
Authored: Mon Mar 16 16:05:28 2015 -0700
Committer: Chris Riccomini <cr...@apache.org>
Committed: Mon Mar 16 16:05:28 2015 -0700
----------------------------------------------------------------------
.../java/org/apache/samza/task/TaskContext.java | 10 ++++
.../apache/samza/container/TaskInstance.scala | 4 ++
.../samza/container/TestTaskInstance.scala | 57 +++++++++++++++++---
3 files changed, 65 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/537262e2/samza-api/src/main/java/org/apache/samza/task/TaskContext.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/task/TaskContext.java b/samza-api/src/main/java/org/apache/samza/task/TaskContext.java
index 6d10212..929409e 100644
--- a/samza-api/src/main/java/org/apache/samza/task/TaskContext.java
+++ b/samza-api/src/main/java/org/apache/samza/task/TaskContext.java
@@ -37,4 +37,14 @@ public interface TaskContext {
Object getStore(String name);
TaskName getTaskName();
+
+ /**
+ * Set the starting offset for the given {@link org.apache.samza.system.SystemStreamPartition}. Offsets
+ * can only be set for a {@link org.apache.samza.system.SystemStreamPartition} assigned to this task
+ * (as returned by {@link #getSystemStreamPartitions()}); trying to set the offset for any other partition
+ * will have no effect.
+ *
+ * NOTE: this feature is experimental, and the API may change in a future release.
+ */
+ void setStartingOffset(SystemStreamPartition ssp, String offset);
}
http://git-wip-us.apache.org/repos/asf/samza/blob/537262e2/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala b/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala
index a583ff9..be0b55a 100644
--- a/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala
+++ b/samza-core/src/main/scala/org/apache/samza/container/TaskInstance.scala
@@ -65,6 +65,10 @@ class TaskInstance(
null
}
def getTaskName = taskName
+
+ override def setStartingOffset(ssp: SystemStreamPartition, offset: String): Unit = {
+ offsetManager.startingOffsets += (ssp -> offset)
+ }
}
def registerMetrics {
http://git-wip-us.apache.org/repos/asf/samza/blob/537262e2/samza-core/src/test/scala/org/apache/samza/container/TestTaskInstance.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/container/TestTaskInstance.scala b/samza-core/src/test/scala/org/apache/samza/container/TestTaskInstance.scala
index 11eab16..54b4df8 100644
--- a/samza-core/src/test/scala/org/apache/samza/container/TestTaskInstance.scala
+++ b/samza-core/src/test/scala/org/apache/samza/container/TestTaskInstance.scala
@@ -40,12 +40,7 @@ import org.apache.samza.system.SystemStreamMetadata
import org.apache.samza.system.SystemStreamMetadata.SystemStreamPartitionMetadata
import org.apache.samza.system.SystemStreamPartition
import org.apache.samza.system.chooser.RoundRobinChooser
-import org.apache.samza.task.MessageCollector
-import org.apache.samza.task.ReadableCoordinator
-import org.apache.samza.task.StreamTask
-import org.apache.samza.task.TaskCoordinator
-import org.apache.samza.task.TaskInstanceCollector
-import org.apache.samza.task.WindowableTask
+import org.apache.samza.task._
import org.junit.Assert._
import org.junit.Test
import org.scalatest.Assertions.intercept
@@ -240,4 +235,54 @@ class TestTaskInstance {
assertEquals(2L, getCount(group, classOf[NonFatalException].getName))
assertEquals(1L, getCount(group, classOf[FatalException].getName))
}
+
+
+ /**
+ * Tests that the init() method of task can override the existing offset
+ * assignment.
+ */
+ @Test
+ def testManualOffsetReset {
+
+ val partition0 = new SystemStreamPartition("system", "stream", new Partition(0))
+ val partition1 = new SystemStreamPartition("system", "stream", new Partition(1))
+
+ val task = new StreamTask with InitableTask {
+
+ override def init(config: Config, context: TaskContext): Unit = {
+
+ assertTrue("Can only update offsets for assigned partition",
+ context.getSystemStreamPartitions.contains(partition1)
+ )
+
+ context.setStartingOffset(partition1, "10")
+ }
+
+ override def process(envelope: IncomingMessageEnvelope, collector: MessageCollector, coordinator: TaskCoordinator): Unit = {}
+ }
+
+ val chooser = new RoundRobinChooser()
+ val consumers = new SystemConsumers(chooser, consumers = Map.empty)
+ val producers = new SystemProducers(Map.empty, new SerdeManager())
+ val collector = new TaskInstanceCollector(producers)
+
+ val offsetManager = new OffsetManager()
+
+ offsetManager.startingOffsets ++= Map(partition0 -> "0", partition1 -> "0")
+
+ val taskInstance = new TaskInstance(
+ task = task,
+ taskName = new TaskName("Offset Reset Task 0"),
+ config = new MapConfig(),
+ metrics = new TaskInstanceMetrics(),
+ consumerMultiplexer = consumers,
+ collector = collector,
+ offsetManager = offsetManager,
+ systemStreamPartitions = Set(partition0, partition1) )
+
+ taskInstance.initTask
+
+ assertEquals(Some("0"), offsetManager.getStartingOffset(partition0))
+ assertEquals(Some("10"), offsetManager.getStartingOffset(partition1))
+ }
}
[28/36] samza git commit: SAMZA-620;
torture test checker should start from oldest offset
Posted by cr...@apache.org.
SAMZA-620; torture test checker should start from oldest offset
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/8aeec8de
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/8aeec8de
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/8aeec8de
Branch: refs/heads/samza-sql
Commit: 8aeec8defcfb62cd805425811520ca6474973126
Parents: d20ae5c
Author: Chris Riccomini <cr...@apache.org>
Authored: Fri Mar 27 11:04:09 2015 -0700
Committer: Chris Riccomini <cr...@apache.org>
Committed: Fri Mar 27 11:04:09 2015 -0700
----------------------------------------------------------------------
samza-test/src/main/config/join/checker.samza | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/8aeec8de/samza-test/src/main/config/join/checker.samza
----------------------------------------------------------------------
diff --git a/samza-test/src/main/config/join/checker.samza b/samza-test/src/main/config/join/checker.samza
index ab557d9..c0e3df7 100644
--- a/samza-test/src/main/config/join/checker.samza
+++ b/samza-test/src/main/config/join/checker.samza
@@ -20,6 +20,7 @@
job.name=checker
systems.kafka.partitioner.class=org.apache.samza.test.integration.join.EpochPartitioner
+systems.kafka.samza.offset.default=oldest
# Task
task.class=org.apache.samza.test.integration.join.Checker
[13/36] samza git commit: SAMZA-602;
fix compilation issue with scala/java dependencies in samza-kafka
Posted by cr...@apache.org.
SAMZA-602; fix compilation issue with scala/java dependencies in samza-kafka
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/7e4f3e49
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/7e4f3e49
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/7e4f3e49
Branch: refs/heads/samza-sql
Commit: 7e4f3e490b1adc2f1ce268f814c26ac0dd0b733e
Parents: 660d879
Author: Chris Riccomini <cr...@apache.org>
Authored: Mon Mar 16 11:19:53 2015 -0700
Committer: Chris Riccomini <cr...@apache.org>
Committed: Mon Mar 16 11:19:53 2015 -0700
----------------------------------------------------------------------
build.gradle | 3 +++
1 file changed, 3 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/7e4f3e49/build.gradle
----------------------------------------------------------------------
diff --git a/build.gradle b/build.gradle
index 87334fa..113da2a 100644
--- a/build.gradle
+++ b/build.gradle
@@ -145,6 +145,9 @@ project(":samza-core_$scalaVersion") {
project(":samza-kafka_$scalaVersion") {
apply plugin: 'scala'
+ sourceSets.test.scala.srcDir "src/test/java"
+ sourceSets.test.java.srcDirs = []
+
configurations {
// Remove transitive dependencies from Zookeeper that we don't want.
compile.exclude group: 'javax.jms', module: 'jms'
[31/36] samza git commit: SAMZA-627: add 0.9.0 release to Archive
category
Posted by cr...@apache.org.
SAMZA-627: add 0.9.0 release to Archive category
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/cde495e1
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/cde495e1
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/cde495e1
Branch: refs/heads/samza-sql
Commit: cde495e161d1a3f29469cb503742da703a0dd4a7
Parents: 95b6afc
Author: Yan Fang <ya...@gmail.com>
Authored: Wed Apr 1 11:39:46 2015 -0700
Committer: Yan Fang <ya...@gmail.com>
Committed: Wed Apr 1 11:39:46 2015 -0700
----------------------------------------------------------------------
docs/_layouts/default.html | 1 +
docs/archive/index.html | 8 ++++++++
2 files changed, 9 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/cde495e1/docs/_layouts/default.html
----------------------------------------------------------------------
diff --git a/docs/_layouts/default.html b/docs/_layouts/default.html
index 13172e7..839aadf 100644
--- a/docs/_layouts/default.html
+++ b/docs/_layouts/default.html
@@ -103,6 +103,7 @@
<h1><i class="fa fa-history"></i> Archive</h1>
<ul>
<li><a href="/archive/index.html#latest">latest</a></li>
+ <li><a href="/archive/index.html#09">0.9</a></li>
<li><a href="/archive/index.html#08">0.8</a></li>
<li><a href="/archive/index.html#07">0.7</a></li>
</ul>
http://git-wip-us.apache.org/repos/asf/samza/blob/cde495e1/docs/archive/index.html
----------------------------------------------------------------------
diff --git a/docs/archive/index.html b/docs/archive/index.html
index 586e352..1bfeb9b 100644
--- a/docs/archive/index.html
+++ b/docs/archive/index.html
@@ -27,6 +27,14 @@ title: Documentation
<li><a href="../startup/hello-samza/latest">Hello Samza</a></li>
</ul>
+<h4 id="09">0.9 Release</h4>
+
+<ul class="documentation-list">
+ <li><a href="../learn/documentation/0.9">Documentation</a></li>
+ <li><a href="../learn/tutorials/0.9">Tutorials</a></li>
+ <li><a href="../startup/hello-samza/0.9">Hello Samza</a></li>
+</ul>
+
<h4 id="08">0.8 Release</h4>
<ul class="documentation-list">
[29/36] samza git commit: SAMZA-621;
fix zopkio test package path for 0.10.0
Posted by cr...@apache.org.
SAMZA-621; fix zopkio test package path for 0.10.0
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/faf93fff
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/faf93fff
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/faf93fff
Branch: refs/heads/samza-sql
Commit: faf93fff3a40b866e5e03fa3f1dde883dd17a926
Parents: 8aeec8d
Author: Chris Riccomini <cr...@apache.org>
Authored: Fri Mar 27 13:48:15 2015 -0700
Committer: Chris Riccomini <cr...@apache.org>
Committed: Fri Mar 27 13:48:15 2015 -0700
----------------------------------------------------------------------
samza-test/src/main/python/configs/tests.json | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/faf93fff/samza-test/src/main/python/configs/tests.json
----------------------------------------------------------------------
diff --git a/samza-test/src/main/python/configs/tests.json b/samza-test/src/main/python/configs/tests.json
index 5251af9..30be820 100644
--- a/samza-test/src/main/python/configs/tests.json
+++ b/samza-test/src/main/python/configs/tests.json
@@ -1,5 +1,5 @@
{
- "samza_executable": "samza-test_2.10-0.9.0-SNAPSHOT.tgz",
+ "samza_executable": "samza-test_2.10-0.10.0-SNAPSHOT.tgz",
"samza_install_path": "deploy/smoke_tests",
"samza_config_factory": "org.apache.samza.config.factories.PropertiesConfigFactory"
}
[35/36] samza git commit: SAMZA-636; add a uuid serde
Posted by cr...@apache.org.
SAMZA-636; add a uuid serde
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/edc2c78f
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/edc2c78f
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/edc2c78f
Branch: refs/heads/samza-sql
Commit: edc2c78f8759bb38dae1c2e0472f1123456f30c5
Parents: b276afe
Author: Mohamed Mahmoud (El-Geish) <el...@gmail.com>
Authored: Tue Apr 7 16:18:35 2015 -0700
Committer: Chris Riccomini <cr...@apache.org>
Committed: Tue Apr 7 16:18:35 2015 -0700
----------------------------------------------------------------------
.../apache/samza/serializers/StringSerde.scala | 1 -
.../apache/samza/serializers/UUIDSerde.scala | 47 +++++++++++++++++
.../samza/serializers/TestIntegerSerde.scala | 3 --
.../samza/serializers/TestUUIDSerde.scala | 53 ++++++++++++++++++++
4 files changed, 100 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/edc2c78f/samza-core/src/main/scala/org/apache/samza/serializers/StringSerde.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/serializers/StringSerde.scala b/samza-core/src/main/scala/org/apache/samza/serializers/StringSerde.scala
index 7b4b23d..6e5f260 100644
--- a/samza-core/src/main/scala/org/apache/samza/serializers/StringSerde.scala
+++ b/samza-core/src/main/scala/org/apache/samza/serializers/StringSerde.scala
@@ -19,7 +19,6 @@
package org.apache.samza.serializers
-import java.nio.ByteBuffer
import org.apache.samza.config.Config
/**
http://git-wip-us.apache.org/repos/asf/samza/blob/edc2c78f/samza-core/src/main/scala/org/apache/samza/serializers/UUIDSerde.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/main/scala/org/apache/samza/serializers/UUIDSerde.scala b/samza-core/src/main/scala/org/apache/samza/serializers/UUIDSerde.scala
new file mode 100644
index 0000000..88d4327
--- /dev/null
+++ b/samza-core/src/main/scala/org/apache/samza/serializers/UUIDSerde.scala
@@ -0,0 +1,47 @@
+/*
+ * 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.samza.serializers
+
+import java.nio.ByteBuffer
+import java.util.UUID
+
+import org.apache.samza.config.Config
+
+/**
+ * A serializer for UUID
+ */
+class UUIDSerdeFactory extends SerdeFactory[UUID] {
+ def getSerde(name: String, config: Config): Serde[UUID] = new UUIDSerde
+}
+
+class UUIDSerde() extends Serde[UUID] {
+ def toBytes(obj: UUID): Array[Byte] = if (obj != null) {
+ ByteBuffer.allocate(16).putLong(obj.getMostSignificantBits).putLong(obj.getLeastSignificantBits).array
+ } else {
+ null
+ }
+
+ def fromBytes(bytes: Array[Byte]): UUID = if (bytes != null) {
+ val buffer = ByteBuffer.wrap(bytes)
+ new UUID(buffer.getLong, buffer.getLong)
+ } else {
+ null
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/edc2c78f/samza-core/src/test/scala/org/apache/samza/serializers/TestIntegerSerde.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/serializers/TestIntegerSerde.scala b/samza-core/src/test/scala/org/apache/samza/serializers/TestIntegerSerde.scala
index ad646d7..a3e7e1f 100644
--- a/samza-core/src/test/scala/org/apache/samza/serializers/TestIntegerSerde.scala
+++ b/samza-core/src/test/scala/org/apache/samza/serializers/TestIntegerSerde.scala
@@ -19,8 +19,6 @@
package org.apache.samza.serializers
-import java.util.Arrays
-
import org.junit.Assert._
import org.junit.Test
@@ -33,7 +31,6 @@ class TestIntegerSerde {
val fooBar = 37
val fooBarBytes = serde.toBytes(fooBar)
- fooBarBytes.foreach(System.err.println)
assertArrayEquals(Array[Byte](0, 0, 0, 37), fooBarBytes)
assertEquals(fooBar, serde.fromBytes(fooBarBytes))
}
http://git-wip-us.apache.org/repos/asf/samza/blob/edc2c78f/samza-core/src/test/scala/org/apache/samza/serializers/TestUUIDSerde.scala
----------------------------------------------------------------------
diff --git a/samza-core/src/test/scala/org/apache/samza/serializers/TestUUIDSerde.scala b/samza-core/src/test/scala/org/apache/samza/serializers/TestUUIDSerde.scala
new file mode 100644
index 0000000..04ddcdb
--- /dev/null
+++ b/samza-core/src/test/scala/org/apache/samza/serializers/TestUUIDSerde.scala
@@ -0,0 +1,53 @@
+/*
+ * 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.samza.serializers
+
+import java.nio.BufferUnderflowException
+import java.util.UUID
+
+import org.junit.Assert._
+import org.junit.Test
+
+class TestUUIDSerde {
+ private val serde = new UUIDSerde
+
+ @Test
+ def testUUIDSerde {
+ val uuid = new UUID(13, 42)
+ val bytes = serde.toBytes(uuid)
+ assertArrayEquals(Array[Byte](0, 0, 0, 0, 0, 0, 0, 13, 0, 0, 0, 0, 0, 0, 0, 42), bytes)
+ assertEquals(uuid, serde.fromBytes(bytes))
+ }
+
+ @Test
+ def testToBytesWhenNull {
+ assertEquals(null, serde.toBytes(null))
+ }
+
+ @Test
+ def testFromBytesWhenNull {
+ assertEquals(null, serde.fromBytes(null))
+ }
+
+ @Test(expected = classOf[BufferUnderflowException])
+ def testFromBytesWhenInvalid {
+ assertEquals(null, serde.fromBytes(Array[Byte](0)))
+ }
+}
[20/36] samza git commit: SAMZA-456;
add samza-yarn and samza-core to multinode yarn tutorial
Posted by cr...@apache.org.
SAMZA-456; add samza-yarn and samza-core to multinode yarn tutorial
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/05a02eed
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/05a02eed
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/05a02eed
Branch: refs/heads/samza-sql
Commit: 05a02eed176ae51d350588260fbd70439ac739c7
Parents: 9a9ab8d
Author: Tommy Becker <tw...@gmail.com>
Authored: Thu Mar 19 09:18:54 2015 -0700
Committer: Chris Riccomini <cr...@apache.org>
Committed: Thu Mar 19 09:18:54 2015 -0700
----------------------------------------------------------------------
.../tutorials/versioned/run-in-multi-node-yarn.md | 14 ++++++++------
1 file changed, 8 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/05a02eed/docs/learn/tutorials/versioned/run-in-multi-node-yarn.md
----------------------------------------------------------------------
diff --git a/docs/learn/tutorials/versioned/run-in-multi-node-yarn.md b/docs/learn/tutorials/versioned/run-in-multi-node-yarn.md
index 969152b..312efaf 100644
--- a/docs/learn/tutorials/versioned/run-in-multi-node-yarn.md
+++ b/docs/learn/tutorials/versioned/run-in-multi-node-yarn.md
@@ -74,16 +74,18 @@ The goal of these steps is to configure YARN to read http filesystem because we
{% highlight bash %}
cd /tmp
-curl http://www.scala-lang.org/files/archive/scala-2.10.3.tgz > scala-2.10.3.tgz
-tar -xvf scala-2.10.3.tgz
+curl http://www.scala-lang.org/files/archive/scala-2.10.4.tgz > scala-2.10.4.tgz
+tar -xvf scala-2.10.4.tgz
{% endhighlight %}
-5\. Add Scala and its log jars.
+5\. Add Scala, its log jars, and Samza's HttpFileSystem implementation.
{% highlight bash %}
-cp /tmp/scala-2.10.3/lib/scala-compiler.jar $HADOOP_YARN_HOME/share/hadoop/hdfs/lib
-cp /tmp/scala-2.10.3/lib/scala-library.jar $HADOOP_YARN_HOME/share/hadoop/hdfs/lib
-curl http://search.maven.org/remotecontent?filepath=org/clapper/grizzled-slf4j_2.10/1.0.1/grizzled-slf4j_2.10-1.0.1.jar > $HADOOP_YARN_HOME/share/hadoop/hdfs/lib/grizzled-slf4j_2.10-1.0.1.jar
+cp /tmp/scala-2.10.4/lib/scala-compiler.jar $HADOOP_YARN_HOME/share/hadoop/hdfs/lib
+cp /tmp/scala-2.10.4/lib/scala-library.jar $HADOOP_YARN_HOME/share/hadoop/hdfs/lib
+curl -L http://search.maven.org/remotecontent?filepath=org/clapper/grizzled-slf4j_2.10/1.0.1/grizzled-slf4j_2.10-1.0.1.jar > $HADOOP_YARN_HOME/share/hadoop/hdfs/lib/grizzled-slf4j_2.10-1.0.1.jar
+curl -L http://search.maven.org/remotecontent?filepath=org/apache/samza/samza-yarn_2.10/0.8.0/samza-yarn_2.10-0.8.0.jar > $HADOOP_YARN_HOME/share/hadoop/hdfs/lib/samza-yarn_2.10-0.8.0.jar
+curl -L http://search.maven.org/remotecontent?filepath=org/apache/samza/samza-core_2.10/0.8.0/samza-core_2.10-0.8.0.jar > $HADOOP_YARN_HOME/share/hadoop/hdfs/lib/samza-core_2.10-0.8.0.jar
{% endhighlight %}
6\. Add http configuration in core-site.xml (create the core-site.xml file and add content).