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/02/12 22:34:54 UTC
samza git commit: SAMZA-522; upgrade to Kafka 0.8.2.0
Repository: samza
Updated Branches:
refs/heads/master eedf2e720 -> 8090d6539
SAMZA-522; upgrade to Kafka 0.8.2.0
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/8090d653
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/8090d653
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/8090d653
Branch: refs/heads/master
Commit: 8090d6539946cd27fa25e0c0d35925c344a2d3ae
Parents: eedf2e7
Author: Chris Riccomini <cr...@apache.org>
Authored: Thu Feb 12 13:34:47 2015 -0800
Committer: Chris Riccomini <cr...@apache.org>
Committed: Thu Feb 12 13:34:47 2015 -0800
----------------------------------------------------------------------
gradle/dependency-versions.gradle | 2 +-
.../samza/checkpoint/kafka/KafkaCheckpointManager.scala | 4 ++--
.../kafka/KafkaCheckpointManagerFactory.scala | 2 +-
.../scala/org/apache/samza/config/KafkaConfig.scala | 12 ++++++++++++
.../apache/samza/system/kafka/KafkaSystemFactory.scala | 2 +-
.../apache/samza/system/kafka/KafkaSystemProducer.scala | 4 ++--
.../apache/samza/system/kafka/MockKafkaProducer.java | 5 +++--
.../samza/system/kafka/TestKafkaSystemAdmin.scala | 4 ++--
.../samza/test/integration/TestStatefulTask.scala | 4 ++--
9 files changed, 26 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/8090d653/gradle/dependency-versions.gradle
----------------------------------------------------------------------
diff --git a/gradle/dependency-versions.gradle b/gradle/dependency-versions.gradle
index 2bb24ad..03c72f8 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-beta"
+ kafkaVersion = "0.8.2.0"
commonsHttpClientVersion = "3.1"
rocksdbVersion = "3.5.1"
yarnVersion = "2.4.0"
http://git-wip-us.apache.org/repos/asf/samza/blob/8090d653/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 c96394d..4a1b31f 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
@@ -61,7 +61,7 @@ class KafkaCheckpointManager(
bufferSize: Int,
fetchSize: Int,
metadataStore: TopicMetadataStore,
- connectProducer: () => Producer,
+ connectProducer: () => Producer[Array[Byte], Array[Byte]],
connectZk: () => ZkClient,
systemStreamPartitionGrouperFactoryString: String,
retryBackoff: ExponentialSleepStrategy = new ExponentialSleepStrategy,
@@ -70,7 +70,7 @@ class KafkaCheckpointManager(
import KafkaCheckpointManager._
var taskNames = Set[TaskName]()
- var producer: Producer = null
+ var producer: Producer[Array[Byte], Array[Byte]] = null
var taskNamesToOffsets: Map[TaskName, Checkpoint] = null
var startingOffset: Option[Long] = None // Where to start reading for each subsequent call of readCheckpoint
http://git-wip-us.apache.org/repos/asf/samza/blob/8090d653/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 0f64c08..7fc6d89 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
@@ -80,7 +80,7 @@ class KafkaCheckpointManagerFactory extends CheckpointManagerFactory with Loggin
val fetchSize = consumerConfig.fetchMessageMaxBytes // must be > buffer size
val connectProducer = () => {
- new KafkaProducer(producerConfig.getProducerProperties)
+ new KafkaProducer[Array[Byte], Array[Byte]](producerConfig.getProducerProperties)
}
val zkConnect = Option(consumerConfig.zkConnect)
.getOrElse(throw new SamzaException("no zookeeper.connect defined in config"))
http://git-wip-us.apache.org/repos/asf/samza/blob/8090d653/samza-kafka/src/main/scala/org/apache/samza/config/KafkaConfig.scala
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/scala/org/apache/samza/config/KafkaConfig.scala b/samza-kafka/src/main/scala/org/apache/samza/config/KafkaConfig.scala
index 6d54350..02a6ea9 100644
--- a/samza-kafka/src/main/scala/org/apache/samza/config/KafkaConfig.scala
+++ b/samza-kafka/src/main/scala/org/apache/samza/config/KafkaConfig.scala
@@ -34,6 +34,7 @@ import java.util
import scala.collection.JavaConverters._
import org.apache.samza.system.kafka.KafkaSystemFactory
import org.apache.samza.config.SystemConfig.Config2System
+import org.apache.kafka.common.serialization.ByteArraySerializer
object KafkaConfig {
val REGEX_RESOLVED_STREAMS = "job.config.rewriter.%s.regex"
@@ -174,9 +175,20 @@ class KafkaProducerConfig(val systemName: String,
val RETRIES_DEFAULT: java.lang.Integer = Integer.MAX_VALUE
def getProducerProperties = {
+ val byteArraySerializerClassName = classOf[ByteArraySerializer].getCanonicalName
val producerProperties: java.util.Map[String, Object] = new util.HashMap[String, Object]()
producerProperties.putAll(properties)
+ if(!producerProperties.containsKey(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG)) {
+ debug("%s undefined. Defaulting to %s." format (ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, byteArraySerializerClassName))
+ producerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, byteArraySerializerClassName)
+ }
+
+ if(!producerProperties.containsKey(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG)) {
+ debug("%s undefined. Defaulting to %s." format (ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, byteArraySerializerClassName))
+ producerProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, byteArraySerializerClassName)
+ }
+
// Always set (new) producer config for max_in_flight_requests_per_connection and retries_config to 1 & INT.MaxValue
// so that the producer does not optimistically send batches asychronously and thereby, messing up the ordering of outgoing messages
// Retries config is set to Max so that when all attempts fail, Samza also fails the send. We do not have any special handler
http://git-wip-us.apache.org/repos/asf/samza/blob/8090d653/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 38e3815..4f15002 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
@@ -67,7 +67,7 @@ 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 getProducer = () => { new KafkaProducer(producerConfig.getProducerProperties) }
+ val getProducer = () => { new KafkaProducer[Array[Byte], Array[Byte]](producerConfig.getProducerProperties) }
val metrics = new KafkaSystemProducerMetrics(systemName, registry)
// Unlike consumer, no need to use encoders here, since they come for free
http://git-wip-us.apache.org/repos/asf/samza/blob/8090d653/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 cd1fced..83668dd 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
@@ -37,11 +37,11 @@ import java.util.concurrent.Future
class KafkaSystemProducer(systemName: String,
retryBackoff: ExponentialSleepStrategy = new ExponentialSleepStrategy,
- getProducer: () => Producer,
+ getProducer: () => Producer[Array[Byte], Array[Byte]],
metrics: KafkaSystemProducerMetrics,
val clock: () => Long = () => System.currentTimeMillis) extends SystemProducer with Logging with TimerUtils
{
- var producer: Producer = null
+ var producer: Producer[Array[Byte], Array[Byte]] = null
val latestFuture: javaMap[String, Future[RecordMetadata]] = new util.HashMap[String, Future[RecordMetadata]]()
val sendFailed: AtomicBoolean = new AtomicBoolean(false)
var exceptionThrown: AtomicReference[Exception] = new AtomicReference[Exception]()
http://git-wip-us.apache.org/repos/asf/samza/blob/8090d653/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
index 3b7d57a..6f498de 100644
--- 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
@@ -40,9 +40,10 @@ 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 {
+public class MockKafkaProducer implements Producer<byte[], byte[]> {
private Cluster _cluster;
private List<FutureTask<RecordMetadata>> _callbacksList = new ArrayList<FutureTask<RecordMetadata>>();
@@ -164,7 +165,7 @@ public class MockKafkaProducer implements Producer {
}
@Override
- public Map<String, ? extends Metric> metrics() {
+ public Map<MetricName, Metric> metrics() {
return null;
}
http://git-wip-us.apache.org/repos/asf/samza/blob/8090d653/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 a055204..3d1e6ec 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
@@ -76,7 +76,7 @@ object TestKafkaSystemAdmin {
config.put("request.required.acks", "-1")
config.put("serializer.class", "kafka.serializer.StringEncoder")
val producerConfig = new KafkaProducerConfig("kafka", "i001", config)
- var producer: KafkaProducer = null
+ var producer: KafkaProducer[Array[Byte], Array[Byte]] = null
var zookeeper: EmbeddedZookeeper = null
var server1: KafkaServer = null
var server2: KafkaServer = null
@@ -90,7 +90,7 @@ object TestKafkaSystemAdmin {
server2 = TestUtils.createServer(new KafkaConfig(props2))
server3 = TestUtils.createServer(new KafkaConfig(props3))
zkClient = new ZkClient(zkConnect + "/", 6000, 6000, ZKStringSerializer)
- producer = new KafkaProducer(producerConfig.getProducerProperties)
+ producer = new KafkaProducer[Array[Byte], Array[Byte]](producerConfig.getProducerProperties)
metadataStore = new ClientUtilTopicMetadataStore(brokers, "some-job-name")
}
http://git-wip-us.apache.org/repos/asf/samza/blob/8090d653/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 4c08d6b..a8b724b 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
@@ -101,7 +101,7 @@ object TestStatefulTask {
config.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, new Integer(1))
config.put(ProducerConfig.RETRIES_CONFIG, new Integer(Integer.MAX_VALUE-1))
val producerConfig = new KafkaProducerConfig("kafka", "i001", config)
- var producer: Producer = null
+ var producer: Producer[Array[Byte], Array[Byte]] = null
val cp1 = new Checkpoint(Map(new SystemStreamPartition("kafka", "topic", new Partition(0)) -> "123"))
val cp2 = new Checkpoint(Map(new SystemStreamPartition("kafka", "topic", new Partition(0)) -> "12345"))
var zookeeper: EmbeddedZookeeper = null
@@ -117,7 +117,7 @@ object TestStatefulTask {
server2 = TestUtils.createServer(new KafkaConfig(props2))
server3 = TestUtils.createServer(new KafkaConfig(props3))
zkClient = new ZkClient(zkConnect + "/", 6000, 6000, ZKStringSerializer)
- producer = new KafkaProducer(producerConfig.getProducerProperties)
+ producer = new KafkaProducer[Array[Byte], Array[Byte]](producerConfig.getProducerProperties)
metadataStore = new ClientUtilTopicMetadataStore(brokers, "some-job-name")
createTopics