You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@eagle.apache.org by qi...@apache.org on 2016/11/10 07:55:24 UTC
incubator-eagle git commit: [EAGLE-761] remove partition key in
kafkaStreamSink
Repository: incubator-eagle
Updated Branches:
refs/heads/master ad8cc9c99 -> 8f8fe57e6
[EAGLE-761] remove partition key in kafkaStreamSink
https://issues.apache.org/jira/browse/EAGLE-761
Author: Zhao, Qingwen <qi...@apache.org>
Closes #643 from qingwen220/EAGLE-761.
Project: http://git-wip-us.apache.org/repos/asf/incubator-eagle/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-eagle/commit/8f8fe57e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-eagle/tree/8f8fe57e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-eagle/diff/8f8fe57e
Branch: refs/heads/master
Commit: 8f8fe57e6ab7484f400333d3369bb849cb2c0e6e
Parents: ad8cc9c
Author: Zhao, Qingwen <qi...@apache.org>
Authored: Thu Nov 10 15:55:13 2016 +0800
Committer: Zhao, Qingwen <qi...@apache.org>
Committed: Thu Nov 10 15:55:13 2016 +0800
----------------------------------------------------------------------
.../main/java/org/apache/eagle/app/sink/KafkaStreamSink.java | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/8f8fe57e/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/KafkaStreamSink.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/KafkaStreamSink.java b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/KafkaStreamSink.java
index 2207c8c..12acb81 100644
--- a/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/KafkaStreamSink.java
+++ b/eagle-core/eagle-app/eagle-app-base/src/main/java/org/apache/eagle/app/sink/KafkaStreamSink.java
@@ -18,13 +18,11 @@ package org.apache.eagle.app.sink;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
-import backtype.storm.topology.BasicOutputCollector;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.typesafe.config.Config;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
-import org.apache.eagle.metadata.utils.StreamIdConversions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -64,7 +62,9 @@ public class KafkaStreamSink extends StormStreamSink<KafkaStreamSinkConfig> {
protected void execute(Object key, Map event, OutputCollector collector) throws Exception {
try {
String output = new ObjectMapper().writeValueAsString(event);
- producer.send(new KeyedMessage(this.topicId, key, output));
+ // partition key may cause data skew
+ //producer.send(new KeyedMessage(this.topicId, key, output));
+ producer.send(new KeyedMessage(this.topicId, output));
} catch (Exception ex) {
LOG.error(ex.getMessage(), ex);
throw ex;