samza git commit: SAMZA-1539: KafkaProducer potential hang on close() when task.drop.pr…
Repository: samza
Updated Branches:
refs/heads/master 29ecae891 -> ef1c9625c
SAMZA-1539: KafkaProducer potential hang on close() when task.drop.prâ¦
â¦oducer.errors==true
Author: Jacob Maes
Reviewers: Boris Shkolnik
Closes #390 from jmakes/samza-1539
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/ef1c9625
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/ef1c9625
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/ef1c9625
Branch: refs/heads/master
Commit: ef1c9625c63b6c169585c5d1928298f36f3037fb
Parents: 29ecae8
Author: Jacob Maes
Authored: Fri Dec 22 10:36:37 2017 -0800
Committer: Jacob Maes <--global>
Committed: Fri Dec 22 10:36:37 2017 -0800
--
.../system/kafka/KafkaSystemProducer.scala | 135 ++-
.../system/kafka/TestKafkaSystemProducer.scala | 39 --
2 files changed, 103 insertions(+), 71 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/samza/blob/ef1c9625/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 5e83666..9eaf895 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
@@ -28,7 +28,6 @@ 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.PartitionInfo
-import org.apache.kafka.common.errors.SerializationException
import org.apache.samza.system.OutgoingMessageEnvelope
import org.apache.samza.system.SystemProducer
import org.apache.samza.system.SystemProducerException
@@ -46,32 +45,30 @@ class KafkaSystemProducer(systemName: String,
// Represents a fatal error that caused the producer to close.
val fatalException: AtomicReference[SystemProducerException] = new
AtomicReference[SystemProducerException]()
- @volatile var producer: Producer[Array[Byte], Array[Byte]] = null
- val producerLock: Object = new Object
+ val producerRef: AtomicReference[Producer[Array[Byte], Array[Byte]]] = new
AtomicReference[Producer[Array[Byte], Array[Byte]]]()
+ val producerCreationLock: Object = new Object
+ @volatile var stopped = false
def start(): Unit = {
-producer = getProducer()
+producerRef.set(getProducer())
}
def stop() {
info("Stopping producer for system: " + this.systemName)
-// stop() should not happen often so no need to optimize locking
-producerLock.synchronized {
- try {
-if (producer != null) {
- producer.close // Also performs the equivalent of a flush()
-}
+stopped = true
+val currentProducer = producerRef.getAndSet(null)
+try {
+ if (currentProducer != null) {
+currentProducer.close // Also performs the equivalent of a flush()
+ }
-val exception = fatalException.get()
-if (exception != null) {
- error("Observed an earlier send() error while closing producer",
exception)
-}
- } catch {
-case e: Exception => error("Error while closing producer for system: "
+ systemName, e)
- } finally {
-producer = null
+ val exception = fatalException.get()
+ if (exception != null) {
+error("Observed an earlier send() error while closing producer",
exception)
}
+} catch {
+ case e: Exception => error("Error while closing producer for system: " +
systemName, e)
}
}
@@ -82,7 +79,7 @@ class KafkaSystemProducer(systemName: String,
trace("Enqueuing message: %s, %s." format (source, envelope))
val topicName = envelope.getSystemStream.getStream
-if (topicName == null || topicName == "") {
+if (topicName == null || topicName.isEmpty) {
throw new IllegalArgumentException("Invalid system stream: " +
envelope.getSystemStream)
}
@@ -92,10 +89,7 @@ class KafkaSystemProducer(systemName: String,
throw new SystemProducerException("Producer was unable to recover from
previous exception.", globalProducerException)
}
-val currentProducer = producer
-if (currentProducer == null) {
- throw new SystemProducerException("Kafka producer is null.")
-}
+val currentProducer = getOrCreateCurrentProducer
// Java-based Kafka producer API requires an "Integer" type partitionKey
and does not allow custom overriding of Partitioners
// Any kind of custom partitioning has to be done on the client-side
@@ -115,7 +109,7 @@ cl
[jira] [Commented] (SAMZA-1539) KafkaProducer potential hang on close() when task.drop.producer.errors==true
[ https://issues.apache.org/jira/browse/SAMZA-1539?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16301783#comment-16301783 ] ASF GitHub Bot commented on SAMZA-1539: --- Github user asfgit closed the pull request at: https://github.com/apache/samza/pull/390 > KafkaProducer potential hang on close() when task.drop.producer.errors==true > > > Key: SAMZA-1539 > URL: https://issues.apache.org/jira/browse/SAMZA-1539 > Project: Samza > Issue Type: Bug >Reporter: Jake Maes >Assignee: Jake Maes > Fix For: 0.15.0 > > > The issue is caused by 2 bad behaviors: > 1. KafkaProducer.close() does an unbounded thread.join() even when a force > close (timeout == 0 ms) is specified. > 2. KafkaSystemProducer.handleSendException() has a synchronized block to > recreate the producer when task.drop.producer.errors==true. When the IO > thread is waiting on the synchronized block, it prevents the KafkaProducer > from joining the IO thread. > Fixing either one will fix the issue, but we will close this ticket as soon > as the Samza side is tested and committed. -- This message was sent by Atlassian JIRA (v6.4.14#64029)
[jira] [Resolved] (SAMZA-1539) KafkaProducer potential hang on close() when task.drop.producer.errors==true
[ https://issues.apache.org/jira/browse/SAMZA-1539?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jake Maes resolved SAMZA-1539. -- Resolution: Fixed Issue resolved by pull request 390 [https://github.com/apache/samza/pull/390] > KafkaProducer potential hang on close() when task.drop.producer.errors==true > > > Key: SAMZA-1539 > URL: https://issues.apache.org/jira/browse/SAMZA-1539 > Project: Samza > Issue Type: Bug >Reporter: Jake Maes >Assignee: Jake Maes > Fix For: 0.15.0 > > > The issue is caused by 2 bad behaviors: > 1. KafkaProducer.close() does an unbounded thread.join() even when a force > close (timeout == 0 ms) is specified. > 2. KafkaSystemProducer.handleSendException() has a synchronized block to > recreate the producer when task.drop.producer.errors==true. When the IO > thread is waiting on the synchronized block, it prevents the KafkaProducer > from joining the IO thread. > Fixing either one will fix the issue, but we will close this ticket as soon > as the Samza side is tested and committed. -- This message was sent by Atlassian JIRA (v6.4.14#64029)
[1/2] samza git commit: Documentation for Samza SQL
Repository: samza
Updated Branches:
refs/heads/0.14.0 ef506a3c5 -> 804aea2b9
http://git-wip-us.apache.org/repos/asf/samza/blob/804aea2b/samza-tools/src/main/java/org/apache/samza/tools/schemas/ProfileChangeEvent.avsc
--
diff --git
a/samza-tools/src/main/java/org/apache/samza/tools/schemas/ProfileChangeEvent.avsc
b/samza-tools/src/main/java/org/apache/samza/tools/schemas/ProfileChangeEvent.avsc
new file mode 100644
index 000..5c1e49d
--- /dev/null
+++
b/samza-tools/src/main/java/org/apache/samza/tools/schemas/ProfileChangeEvent.avsc
@@ -0,0 +1,51 @@
+/*
+* 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.
+*/
+
+{
+"name": "ProfileChangeEvent",
+"version" : 1,
+"namespace": "com.linkedin.samza.tools.avro",
+"type": "record",
+"fields": [
+{
+"name": "Name",
+"doc": "Name of the profile.",
+"type": ["null", "string"],
+"default":null
+},
+{
+"name": "NewCompany",
+"doc": "Name of the new company the person joined.",
+"type": ["null", "string"],
+"default":null
+},
+{
+"name": "OldCompany",
+"doc": "Name of the old company the person was working.",
+"type": ["null", "string"],
+"default":null
+},
+{
+"name": "ProfileChangeTimestamp",
+"doc": "Time at which the profile was changed.",
+"type": ["null", "long"],
+"default":null
+}
+]
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/804aea2b/samza-tools/src/main/java/org/apache/samza/tools/schemas/ProfileChangeEvent.java
--
diff --git
a/samza-tools/src/main/java/org/apache/samza/tools/schemas/ProfileChangeEvent.java
b/samza-tools/src/main/java/org/apache/samza/tools/schemas/ProfileChangeEvent.java
new file mode 100644
index 000..c8e951c
--- /dev/null
+++
b/samza-tools/src/main/java/org/apache/samza/tools/schemas/ProfileChangeEvent.java
@@ -0,0 +1,60 @@
+/*
+* 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.
+*/
+
+/**
+ * Autogenerated by Avro
+ *
+ * DO NOT EDIT DIRECTLY
+ */
+package org.apache.samza.tools.schemas;
+
+@SuppressWarnings("all")
+public class ProfileChangeEvent extends
org.apache.avro.specific.SpecificRecordBase implements
org.apache.avro.specific.SpecificRecord {
+ public static final org.apache.avro.Schema SCHEMA$ =
org.apache.avro.Schema.parse("{\"type\":\"record\",\"name\":\"ProfileChangeEvent\",\"namespace\":\"com.linkedin.samza.tools.avro\",\"fields\":[{\"name\":\"Name\",\"type\":[\"null\",\"string\"],\"doc\":\"Name
of the
profile.\",\"default\":null},{\"name\":\"NewCompany\",\"type\":[\"null\",\"string\"],\"doc\":\"Name
of the new company the person
joined.\",\"default\":null},{\"name\":\"OldCompany\",\"type\":[\"null\",\"string\"],\"doc\":\"Name
of the old company the person was
working.\",\"default\":null},{\"name\":\"ProfileChangeTimestamp\",\"type\":[\"null\",\"long\"],\"doc\":\"Time
at which the profile was changed.\",\"default\":null}]}");
+ /** Name of the profile. */
+ public java.lang.CharSequence Name;
+ /** Name of the new company the person joined. */
+ public java.lang.CharSequence NewCompany;
+ /** Name of the old company the person was working. */
+ public java.lang.CharSequence OldCompany;
+ /** Time at whi
[2/2] samza git commit: Documentation for Samza SQL
Documentation for Samza SQL
**Samza tools** :
Contains the following tools that can be used for playing with Samza sql or any
other samza job.
1. Generate kafka events : Tool used to generate avro serialized kafka events
2. Event hub consumer : Tool used to consume events from event hubs topic. This
can be used if the samza job writes events to event hubs.
3. Samza sql console : Tool used to execute SQL using samza sql.
Adds documentation on how to use Samza SQL on a local machine and on a yarn
environment and their associated Samza tooling.
https://issues.apache.org/jira/browse/SAMZA-1526
Author: Srinivasulu Punuru
Reviewers: Yi Pan, Jagadish
Closes #374 from srinipunuru/docs.1
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/804aea2b
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/804aea2b
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/804aea2b
Branch: refs/heads/0.14.0
Commit: 804aea2b957fbd30a2b79d5a07fbb168e2871244
Parents: ef506a3
Author: Srinivasulu Punuru
Authored: Tue Dec 12 10:46:01 2017 -0800
Committer: xiliu
Committed: Fri Dec 22 10:48:45 2017 -0800
--
build.gradle| 28 +++
docs/README.md | 4 +-
docs/learn/tutorials/versioned/index.md | 3 +
docs/learn/tutorials/versioned/samza-sql.md | 123 +++
docs/learn/tutorials/versioned/samza-tools.md | 109 ++
docs/startup/download/index.md | 7 +
gradle/dependency-versions.gradle | 1 +
.../apache/samza/sql/avro/AvroRelConverter.java | 6 +-
samza-tools/config/eh-consumer-log4j.xml| 35
.../config/generate-kafka-events-log4j.xml | 35
samza-tools/config/samza-sql-console-log4j.xml | 35
samza-tools/scripts/eh-consumer.sh | 34 +++
samza-tools/scripts/generate-kafka-events.sh| 34 +++
samza-tools/scripts/samza-sql-console.sh| 34 +++
.../apache/samza/tools/CommandLineHelper.java | 42
.../tools/ConsoleLoggingSystemFactory.java | 126
.../samza/tools/EventHubConsoleConsumer.java| 120 +++
.../apache/samza/tools/GenerateKafkaEvents.java | 205 +++
.../samza/tools/RandomValueGenerator.java | 87
.../org/apache/samza/tools/SamzaSqlConsole.java | 188 +
.../tools/avro/AvroSchemaGenRelConverter.java | 94 +
.../avro/AvroSchemaGenRelConverterFactory.java | 43
.../samza/tools/avro/AvroSerDeFactory.java | 96 +
.../tools/json/JsonRelConverterFactory.java | 93 +
.../samza/tools/schemas/PageViewEvent.avsc | 51 +
.../samza/tools/schemas/PageViewEvent.java | 60 ++
.../samza/tools/schemas/ProfileChangeEvent.avsc | 51 +
.../samza/tools/schemas/ProfileChangeEvent.java | 60 ++
.../apache/samza/tools/udf/RegexMatchUdf.java | 40
samza-tools/src/main/resources/log4j.xml| 43
settings.gradle | 3 +-
31 files changed, 1886 insertions(+), 4 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/samza/blob/804aea2b/build.gradle
--
diff --git a/build.gradle b/build.gradle
index 50cc5e0..860316e 100644
--- a/build.gradle
+++ b/build.gradle
@@ -328,6 +328,34 @@ project(':samza-sql') {
}
}
+project(':samza-tools') {
+ apply plugin: 'java'
+
+ dependencies {
+compile project(':samza-sql')
+compile project(':samza-api')
+compile project(':samza-azure')
+compile "log4j:log4j:$log4jVersion"
+compile "org.slf4j:slf4j-api:$slf4jVersion"
+compile "org.slf4j:slf4j-log4j12:$slf4jVersion"
+compile "commons-cli:commons-cli:$commonsCliVersion"
+compile "org.apache.avro:avro:$avroVersion"
+compile "org.apache.commons:commons-lang3:$commonsLang3Version"
+compile "org.apache.kafka:kafka-clients:$kafkaVersion"
+ }
+
+ tasks.create(name: "releaseToolsTarGz", dependsOn:
configurations.archives.artifacts, type: Tar) {
+into "samza-tools-${version}"
+compression = Compression.GZIP
+from(project.file("./scripts")) { into "scripts/" }
+from(project.file("./config")) { into "config/" }
+from(project(':samza-shell').file("src/main/bash/run-class.sh")) { into
"scripts/" }
+from(configurations.runtime) { into("lib/") }
+from(configurations.archives.artifacts.files) { into("lib/") }
+duplicatesStrategy 'exclude'
+ }
+}
+
project(":samza-kafka_$scalaVersion") {
apply plugin: 'scala'
http://git-wip-us.apache.org/repos/asf/samza/blob/804aea2b/docs/README.md
--
diff --git a/docs/README.md b/docs/README.md
index 21a4991..3de2a78 1006
samza git commit: Fix a couple sonarcloud issues with samza-1537
Repository: samza
Updated Branches:
refs/heads/master ef1c9625c -> 9a8099cda
Fix a couple sonarcloud issues with samza-1537
Author: Jacob Maes
Reviewers: Jagadish
Closes #393 from jmakes/streamappender-sonarcloud
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/9a8099cd
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/9a8099cd
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/9a8099cd
Branch: refs/heads/master
Commit: 9a8099cdaa2a8ee6fa64d263ebf6704ee5a108da
Parents: ef1c962
Author: Jacob Maes
Authored: Fri Dec 22 11:41:27 2017 -0800
Committer: Jacob Maes <--global>
Committed: Fri Dec 22 11:41:27 2017 -0800
--
.../java/org/apache/samza/logging/log4j/StreamAppender.java | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/samza/blob/9a8099cd/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 0ea8b68..5f41959 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
@@ -152,7 +152,7 @@ public class StreamAppender extends AppenderSkeleton {
// Emit a metric which can be monitored to ensure it doesn't
happen often.
metrics.logMessagesDropped.inc(messagesDropped);
}
- metrics.bufferFillPct.set(Math.round(100 * logQueue.size() /
DEFAULT_QUEUE_SIZE));
+ metrics.bufferFillPct.set(Math.round(100f * logQueue.size() /
DEFAULT_QUEUE_SIZE));
}
} catch (Exception e) {
System.err.println("[StreamAppender] Error sending log message:");
@@ -188,7 +188,8 @@ public class StreamAppender extends AppenderSkeleton {
try {
transferThread.join();
} catch (InterruptedException e) {
-log.error("Interrupted while waiting for sink thread to finish.", e);
+log.error("Interrupted while waiting for transfer thread to finish.",
e);
+Thread.currentThread().interrupt();
}
flushSystemProducer();
[samza] Git Push Summary
Repository: samza Updated Tags: refs/tags/release-0.14.0-rc4 [created] c0ffae6f6
[jira] [Created] (SAMZA-1547) Fix default value for the grouper-factory configuration in KafkaCheckpointManager
Jagadish created SAMZA-1547: --- Summary: Fix default value for the grouper-factory configuration in KafkaCheckpointManager Key: SAMZA-1547 URL: https://issues.apache.org/jira/browse/SAMZA-1547 Project: Samza Issue Type: Bug Reporter: Jagadish -- This message was sent by Atlassian JIRA (v6.4.14#64029)
[jira] [Commented] (SAMZA-1547) Fix default value for the grouper-factory configuration in KafkaCheckpointManager
[ https://issues.apache.org/jira/browse/SAMZA-1547?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16301951#comment-16301951 ] ASF GitHub Bot commented on SAMZA-1547: --- GitHub user vjagadish1989 opened a pull request: https://github.com/apache/samza/pull/394 SAMZA-1547: Parse default value grouper-factory config in KafkaCheckpointMgr - Additionally, updated unit-tests. You can merge this pull request into a Git repository by running: $ git pull https://github.com/vjagadish1989/samza kcm-fix Alternatively you can review and apply these changes as the patch at: https://github.com/apache/samza/pull/394.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #394 commit 3a08518509c311397e23692e9bb1bad037c96a68 Author: Jagadish Date: 2017-12-22T20:49:50Z SAMZA-1547: Fix default handling of grouper-factory configuration in KafkaCheckpointManager > Fix default value for the grouper-factory configuration in > KafkaCheckpointManager > - > > Key: SAMZA-1547 > URL: https://issues.apache.org/jira/browse/SAMZA-1547 > Project: Samza > Issue Type: Bug >Reporter: Jagadish > -- This message was sent by Atlassian JIRA (v6.4.14#64029)
samza git commit: SAMZA-1547; Parse default value grouper-factory config in KafkaCheckpointMgr
Repository: samza
Updated Branches:
refs/heads/master 9a8099cda -> 93219c78d
SAMZA-1547; Parse default value grouper-factory config in KafkaCheckpointMgr
- Additionally, updated all unit-tests.
Author: Jagadish
Reviewers: Prateek M
Closes #394 from vjagadish1989/kcm-fix
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/93219c78
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/93219c78
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/93219c78
Branch: refs/heads/master
Commit: 93219c78d6f473d2c7ad4d8702a25f704ee61a7a
Parents: 9a8099c
Author: Jagadish
Authored: Fri Dec 22 13:58:33 2017 -0800
Committer: Jagadish
Committed: Fri Dec 22 13:58:33 2017 -0800
--
.../org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala | 2 +-
.../apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.scala | 1 -
2 files changed, 1 insertion(+), 2 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/samza/blob/93219c78/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 217b2b6..e1187c5 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
@@ -59,7 +59,7 @@ class KafkaCheckpointManager(checkpointSpec: KafkaStreamSpec,
val checkpointTopic: String = checkpointSpec.getPhysicalName
val checkpointSsp = new SystemStreamPartition(checkpointSystem,
checkpointTopic, new Partition(0))
val checkpointKeySerde = new KafkaCheckpointLogKeySerde
- val expectedGrouperFactory = config.get(JobConfig.SSP_GROUPER_FACTORY)
+ val expectedGrouperFactory = new
JobConfig(config).getSystemStreamPartitionGrouperFactory
val systemProducer = systemFactory.getProducer(checkpointSystem, config,
metricsRegistry)
val systemConsumer = systemFactory.getConsumer(checkpointSystem, config,
metricsRegistry)
http://git-wip-us.apache.org/repos/asf/samza/blob/93219c78/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.scala
--
diff --git
a/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.scala
b/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.scala
index dcf4068..ec9f3a0 100644
---
a/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.scala
+++
b/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.scala
@@ -150,7 +150,6 @@ class TestKafkaCheckpointManager extends
KafkaServerTestHarness {
new MapConfig(new ImmutableMap.Builder[String, String]()
.put(JobConfig.JOB_NAME, "some-job-name")
.put(JobConfig.JOB_ID, "i001")
- .put(JobConfig.SSP_GROUPER_FACTORY, sspGrouperFactoryName)
.put(s"systems.$checkpointSystemName.samza.factory",
classOf[KafkaSystemFactory].getCanonicalName)
.put(s"systems.$checkpointSystemName.producer.bootstrap.servers",
brokers)
.put(s"systems.$checkpointSystemName.consumer.zookeeper.connect",
zkConnect)
samza git commit: SAMZA-1547; Parse default value grouper-factory config in KafkaCheckpointMgr
Repository: samza
Updated Branches:
refs/heads/0.14.0 804aea2b9 -> 7b42474d3
SAMZA-1547; Parse default value grouper-factory config in KafkaCheckpointMgr
- Additionally, updated all unit-tests.
Author: Jagadish
Reviewers: Prateek M
Closes #394 from vjagadish1989/kcm-fix
(cherry picked from commit 93219c78d6f473d2c7ad4d8702a25f704ee61a7a)
Signed-off-by: Jagadish
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/7b42474d
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/7b42474d
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/7b42474d
Branch: refs/heads/0.14.0
Commit: 7b42474d33f96eebb81ad2542d0f2297d1d51d34
Parents: 804aea2
Author: Jagadish
Authored: Fri Dec 22 13:58:33 2017 -0800
Committer: Jagadish
Committed: Fri Dec 22 13:59:31 2017 -0800
--
.../org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala | 2 +-
.../apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.scala | 1 -
2 files changed, 1 insertion(+), 2 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/samza/blob/7b42474d/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 217b2b6..e1187c5 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
@@ -59,7 +59,7 @@ class KafkaCheckpointManager(checkpointSpec: KafkaStreamSpec,
val checkpointTopic: String = checkpointSpec.getPhysicalName
val checkpointSsp = new SystemStreamPartition(checkpointSystem,
checkpointTopic, new Partition(0))
val checkpointKeySerde = new KafkaCheckpointLogKeySerde
- val expectedGrouperFactory = config.get(JobConfig.SSP_GROUPER_FACTORY)
+ val expectedGrouperFactory = new
JobConfig(config).getSystemStreamPartitionGrouperFactory
val systemProducer = systemFactory.getProducer(checkpointSystem, config,
metricsRegistry)
val systemConsumer = systemFactory.getConsumer(checkpointSystem, config,
metricsRegistry)
http://git-wip-us.apache.org/repos/asf/samza/blob/7b42474d/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.scala
--
diff --git
a/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.scala
b/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.scala
index dcf4068..ec9f3a0 100644
---
a/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.scala
+++
b/samza-kafka/src/test/scala/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.scala
@@ -150,7 +150,6 @@ class TestKafkaCheckpointManager extends
KafkaServerTestHarness {
new MapConfig(new ImmutableMap.Builder[String, String]()
.put(JobConfig.JOB_NAME, "some-job-name")
.put(JobConfig.JOB_ID, "i001")
- .put(JobConfig.SSP_GROUPER_FACTORY, sspGrouperFactoryName)
.put(s"systems.$checkpointSystemName.samza.factory",
classOf[KafkaSystemFactory].getCanonicalName)
.put(s"systems.$checkpointSystemName.producer.bootstrap.servers",
brokers)
.put(s"systems.$checkpointSystemName.consumer.zookeeper.connect",
zkConnect)
[jira] [Commented] (SAMZA-1547) Fix default value for the grouper-factory configuration in KafkaCheckpointManager
[ https://issues.apache.org/jira/browse/SAMZA-1547?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16301990#comment-16301990 ] ASF GitHub Bot commented on SAMZA-1547: --- Github user asfgit closed the pull request at: https://github.com/apache/samza/pull/394 > Fix default value for the grouper-factory configuration in > KafkaCheckpointManager > - > > Key: SAMZA-1547 > URL: https://issues.apache.org/jira/browse/SAMZA-1547 > Project: Samza > Issue Type: Bug >Reporter: Jagadish > -- This message was sent by Atlassian JIRA (v6.4.14#64029)
[samza] Git Push Summary
Repository: samza Updated Tags: refs/tags/release-0.14.0-rc5 [created] d49606bfc
