You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2017/12/14 20:22:38 UTC
nifi git commit: NIFI-4675 Lifted restriction on demarcator and
kafka.key usage together. This closes #2326.
Repository: nifi
Updated Branches:
refs/heads/master 48ae4be01 -> d543cfde2
NIFI-4675 Lifted restriction on demarcator and kafka.key usage together. This closes #2326.
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/d543cfde
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/d543cfde
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/d543cfde
Branch: refs/heads/master
Commit: d543cfde2552b6ac5f66d3c723afe60142f3d625
Parents: 48ae4be
Author: jknulst <jk...@hortonworks.com>
Authored: Wed Dec 6 21:31:57 2017 +0100
Committer: Mark Payne <ma...@hotmail.com>
Committed: Thu Dec 14 15:15:13 2017 -0500
----------------------------------------------------------------------
.../nifi/processors/kafka/pubsub/PublishKafka_0_10.java | 9 ++++-----
.../apache/nifi/processors/kafka/pubsub/PublisherLease.java | 5 +----
.../nifi/processors/kafka/pubsub/PublishKafka_0_11.java | 9 ++++-----
.../apache/nifi/processors/kafka/pubsub/PublisherLease.java | 5 +----
.../apache/nifi/processors/kafka/pubsub/PublishKafka.java | 9 ++++-----
.../apache/nifi/processors/kafka/pubsub/PublisherLease.java | 5 +----
.../nifi/processors/kafka/pubsub/PublishKafka_1_0.java | 9 ++++-----
.../apache/nifi/processors/kafka/pubsub/PublisherLease.java | 5 +----
8 files changed, 20 insertions(+), 36 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/d543cfde/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafka_0_10.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafka_0_10.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafka_0_10.java
index 1c7777f..c9d0f37 100644
--- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafka_0_10.java
+++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafka_0_10.java
@@ -148,8 +148,10 @@ public class PublishKafka_0_10 extends AbstractProcessor {
.name("kafka-key")
.displayName("Kafka Key")
.description("The Key to use for the Message. "
- + "If not specified, the flow file attribute 'kafka.key' is used as the message key, if it is present "
- + "and we're not demarcating.")
+ + "If not specified, the flow file attribute 'kafka.key' is used as the message key, if it is present."
+ + "Beware that setting Kafka key and demarcating at the same time may potentially lead to many Kafka messages with the same key."
+ + "Normally this is not a problem as Kafka does not enforce or assume message and key uniqueness. Still, setting the demarcator and Kafka key at the same time poses a risk of "
+ + "data loss on Kafka. During a topic compaction on Kafka, messages will be deduplicated based on this key.")
.required(false)
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.expressionLanguageSupported(true)
@@ -370,9 +372,6 @@ public class PublishKafka_0_10 extends AbstractProcessor {
private byte[] getMessageKey(final FlowFile flowFile, final ProcessContext context) {
- if (context.getProperty(MESSAGE_DEMARCATOR).isSet()) {
- return null;
- }
final String uninterpretedKey;
if (context.getProperty(KEY).isSet()) {
http://git-wip-us.apache.org/repos/asf/nifi/blob/d543cfde/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublisherLease.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublisherLease.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublisherLease.java
index 5a05a08..c17c331 100644
--- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublisherLease.java
+++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublisherLease.java
@@ -75,10 +75,7 @@ public class PublisherLease implements Closeable {
byte[] messageContent;
try {
while ((messageContent = demarcator.nextToken()) != null) {
- // We do not want to use any key if we have a demarcator because that would result in
- // the key being the same for multiple messages
- final byte[] keyToUse = demarcatorBytes == null ? messageKey : null;
- publish(flowFile, keyToUse, messageContent, topic, tracker);
+ publish(flowFile, messageKey, messageContent, topic, tracker);
if (tracker.isFailed(flowFile)) {
// If we have a failure, don't try to send anything else.
http://git-wip-us.apache.org/repos/asf/nifi/blob/d543cfde/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-11-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafka_0_11.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-11-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafka_0_11.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-11-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafka_0_11.java
index ab6b84c..130954a 100644
--- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-11-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafka_0_11.java
+++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-11-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafka_0_11.java
@@ -150,8 +150,10 @@ public class PublishKafka_0_11 extends AbstractProcessor {
.name("kafka-key")
.displayName("Kafka Key")
.description("The Key to use for the Message. "
- + "If not specified, the flow file attribute 'kafka.key' is used as the message key, if it is present "
- + "and we're not demarcating.")
+ + "If not specified, the flow file attribute 'kafka.key' is used as the message key, if it is present."
+ + "Beware that setting Kafka key and demarcating at the same time may potentially lead to many Kafka messages with the same key."
+ + "Normally this is not a problem as Kafka does not enforce or assume message and key uniqueness. Still, setting the demarcator and Kafka key at the same time poses a risk of "
+ + "data loss on Kafka. During a topic compaction on Kafka, messages will be deduplicated based on this key.")
.required(false)
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.expressionLanguageSupported(true)
@@ -433,9 +435,6 @@ public class PublishKafka_0_11 extends AbstractProcessor {
private byte[] getMessageKey(final FlowFile flowFile, final ProcessContext context) {
- if (context.getProperty(MESSAGE_DEMARCATOR).isSet()) {
- return null;
- }
final String uninterpretedKey;
if (context.getProperty(KEY).isSet()) {
http://git-wip-us.apache.org/repos/asf/nifi/blob/d543cfde/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-11-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublisherLease.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-11-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublisherLease.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-11-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublisherLease.java
index d18df7f..e8d744a 100644
--- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-11-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublisherLease.java
+++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-11-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublisherLease.java
@@ -117,10 +117,7 @@ public class PublisherLease implements Closeable {
byte[] messageContent;
try {
while ((messageContent = demarcator.nextToken()) != null) {
- // We do not want to use any key if we have a demarcator because that would result in
- // the key being the same for multiple messages
- final byte[] keyToUse = demarcatorBytes == null ? messageKey : null;
- publish(flowFile, keyToUse, messageContent, topic, tracker);
+ publish(flowFile, messageKey, messageContent, topic, tracker);
if (tracker.isFailed(flowFile)) {
// If we have a failure, don't try to send anything else.
http://git-wip-us.apache.org/repos/asf/nifi/blob/d543cfde/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafka.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafka.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafka.java
index af171bb..32d1ea7 100644
--- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafka.java
+++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafka.java
@@ -150,8 +150,10 @@ public class PublishKafka extends AbstractProcessor {
.name("kafka-key")
.displayName("Kafka Key")
.description("The Key to use for the Message. "
- + "If not specified, the flow file attribute 'kafka.key' is used as the message key, if it is present "
- + "and we're not demarcating.")
+ + "If not specified, the flow file attribute 'kafka.key' is used as the message key, if it is present."
+ + "Beware that setting Kafka key and demarcating at the same time may potentially lead to many Kafka messages with the same key."
+ + "Normally this is not a problem as Kafka does not enforce or assume message and key uniqueness. Still, setting the demarcator and Kafka key at the same time poses a risk of "
+ + "data loss on Kafka. During a topic compaction on Kafka, messages will be deduplicated based on this key.")
.required(false)
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.expressionLanguageSupported(true)
@@ -372,9 +374,6 @@ public class PublishKafka extends AbstractProcessor {
private byte[] getMessageKey(final FlowFile flowFile, final ProcessContext context) {
- if (context.getProperty(MESSAGE_DEMARCATOR).isSet()) {
- return null;
- }
final String uninterpretedKey;
if (context.getProperty(KEY).isSet()) {
http://git-wip-us.apache.org/repos/asf/nifi/blob/d543cfde/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublisherLease.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublisherLease.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublisherLease.java
index b67e8a8..8fb4e67 100644
--- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublisherLease.java
+++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublisherLease.java
@@ -65,10 +65,7 @@ public class PublisherLease implements Closeable {
byte[] messageContent;
try {
while ((messageContent = demarcator.nextToken()) != null) {
- // We do not want to use any key if we have a demarcator because that would result in
- // the key being the same for multiple messages
- final byte[] keyToUse = demarcatorBytes == null ? messageKey : null;
- publish(flowFile, keyToUse, messageContent, topic, tracker);
+ publish(flowFile, messageKey, messageContent, topic, tracker);
if (tracker.isFailed(flowFile)) {
// If we have a failure, don't try to send anything else.
http://git-wip-us.apache.org/repos/asf/nifi/blob/d543cfde/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-1-0-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafka_1_0.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-1-0-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafka_1_0.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-1-0-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafka_1_0.java
index 6f17bd5..48f7747 100644
--- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-1-0-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafka_1_0.java
+++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-1-0-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafka_1_0.java
@@ -150,8 +150,10 @@ public class PublishKafka_1_0 extends AbstractProcessor {
.name("kafka-key")
.displayName("Kafka Key")
.description("The Key to use for the Message. "
- + "If not specified, the flow file attribute 'kafka.key' is used as the message key, if it is present "
- + "and we're not demarcating.")
+ + "If not specified, the flow file attribute 'kafka.key' is used as the message key, if it is present."
+ + "Beware that setting Kafka key and demarcating at the same time may potentially lead to many Kafka messages with the same key."
+ + "Normally this is not a problem as Kafka does not enforce or assume message and key uniqueness. Still, setting the demarcator and Kafka key at the same time poses a risk of "
+ + "data loss on Kafka. During a topic compaction on Kafka, messages will be deduplicated based on this key.")
.required(false)
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.expressionLanguageSupported(true)
@@ -433,9 +435,6 @@ public class PublishKafka_1_0 extends AbstractProcessor {
private byte[] getMessageKey(final FlowFile flowFile, final ProcessContext context) {
- if (context.getProperty(MESSAGE_DEMARCATOR).isSet()) {
- return null;
- }
final String uninterpretedKey;
if (context.getProperty(KEY).isSet()) {
http://git-wip-us.apache.org/repos/asf/nifi/blob/d543cfde/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-1-0-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublisherLease.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-1-0-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublisherLease.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-1-0-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublisherLease.java
index d18df7f..e8d744a 100644
--- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-1-0-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublisherLease.java
+++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-1-0-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublisherLease.java
@@ -117,10 +117,7 @@ public class PublisherLease implements Closeable {
byte[] messageContent;
try {
while ((messageContent = demarcator.nextToken()) != null) {
- // We do not want to use any key if we have a demarcator because that would result in
- // the key being the same for multiple messages
- final byte[] keyToUse = demarcatorBytes == null ? messageKey : null;
- publish(flowFile, keyToUse, messageContent, topic, tracker);
+ publish(flowFile, messageKey, messageContent, topic, tracker);
if (tracker.isFailed(flowFile)) {
// If we have a failure, don't try to send anything else.