You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by GitBox <gi...@apache.org> on 2021/03/23 10:34:56 UTC
[GitHub] [nifi-minifi-cpp] fgerlits opened a new pull request #1037: MINIFICPP-1520 Fix properties to support expression language
fgerlits opened a new pull request #1037:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1037
https://issues.apache.org/jira/browse/MINIFICPP-1520
Contains the following fixes:
* Known Brokers, Client Name, Delivery Guarantee now support expression language
* Request Timeout and Message Timeout no longer claim to support expression language (as this is incompatible with TimePeriodValue validation)
* deprecate Messge Key Field, which didn't work, and replace it with Kafka Key (which is what this property is called in NiFi); Kafka Key supports expression language
* dynamic properties support expression language now
Note: the unit test is not great, as it requires a running Kafka instance on the host, and it looks like it's included in CTest but isn't really. However, this is better than nothing, it's the same as the other two existing Kafka unit tests, and I have opened a Jira to fix these problems: https://issues.apache.org/jira/browse/MINIFICPP-1530
---
Thank you for submitting a contribution to Apache NiFi - MiNiFi C++.
In order to streamline the review of the contribution we ask you
to ensure the following steps have been taken:
### For all changes:
- [x] Is there a JIRA ticket associated with this PR? Is it referenced
in the commit message?
- [x] Does your PR title start with MINIFICPP-XXXX where XXXX is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character.
- [x] Has your PR been rebased against the latest commit within the target branch (typically main)?
- [ ] Is your initial contribution a single, squashed commit?
### For code changes:
- [ ] If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under [ASF 2.0](http://www.apache.org/legal/resolved.html#category-a)?
- [ ] If applicable, have you updated the LICENSE file?
- [ ] If applicable, have you updated the NOTICE file?
### For documentation related changes:
- [ ] Have you ensured that format looks appropriate for the output in which it is rendered?
### Note:
Please ensure that once the PR is submitted, you check GitHub Actions CI results for build issues and submit an update to your PR as soon as possible.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] arpadboda commented on a change in pull request #1037: MINIFICPP-1520 Fix PublishKafka properties to support expression language
Posted by GitBox <gi...@apache.org>.
arpadboda commented on a change in pull request #1037:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1037#discussion_r600527153
##########
File path: extensions/librdkafka/PublishKafka.cpp
##########
@@ -541,6 +543,11 @@ void PublishKafka::onSchedule(const std::shared_ptr<core::ProcessContext> &conte
conn_ = utils::make_unique<KafkaConnection>(key_);
configureNewConnection(context);
+ std::string message_key_field;
+ if (context->getProperty(MessageKeyField.getName(), message_key_field)) {
+ logger_->log_error("The %s property is set. This property is DEPRECATED and has no effect; please use Kafka Key instead.", MessageKeyField.getName());
Review comment:
I tend to agree with the error approach. Especially as this is only going to be logged once, so won't spam the log with error messages.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] szaszm commented on a change in pull request #1037: MINIFICPP-1520 Fix PublishKafka properties to support expression language
Posted by GitBox <gi...@apache.org>.
szaszm commented on a change in pull request #1037:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1037#discussion_r600542921
##########
File path: extensions/librdkafka/PublishKafka.cpp
##########
@@ -541,6 +543,11 @@ void PublishKafka::onSchedule(const std::shared_ptr<core::ProcessContext> &conte
conn_ = utils::make_unique<KafkaConnection>(key_);
configureNewConnection(context);
+ std::string message_key_field;
+ if (context->getProperty(MessageKeyField.getName(), message_key_field)) {
+ logger_->log_error("The %s property is set. This property is DEPRECATED and has no effect; please use Kafka Key instead.", MessageKeyField.getName());
Review comment:
ok
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] fgerlits commented on a change in pull request #1037: MINIFICPP-1520 Fix PublishKafka properties to support expression language
Posted by GitBox <gi...@apache.org>.
fgerlits commented on a change in pull request #1037:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1037#discussion_r600526296
##########
File path: extensions/librdkafka/tests/CMakeLists.txt
##########
@@ -29,10 +29,12 @@ FOREACH(testfile ${KAFKA_TESTS})
createTests("${testfilename}")
MATH(EXPR KAFKA_TEST_COUNT "${KAFKA_TEST_COUNT}+1")
# The line below handles integration test
+ target_include_directories(${testfilename} BEFORE PRIVATE "../..")
Review comment:
I have removed the comment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #1037: MINIFICPP-1520 Fix PublishKafka properties to support expression language
Posted by GitBox <gi...@apache.org>.
adamdebreceni commented on a change in pull request #1037:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1037#discussion_r599563332
##########
File path: extensions/librdkafka/PublishKafka.cpp
##########
@@ -925,12 +931,10 @@ void PublishKafka::onTrigger(const std::shared_ptr<core::ProcessContext> &contex
}
std::string kafkaKey;
- kafkaKey = "";
- if (context->getDynamicProperty(MessageKeyField, kafkaKey, flowFile) && !kafkaKey.empty()) {
- logger_->log_debug("PublishKafka: Message Key Field [%s]", kafkaKey);
- } else {
+ if (!context->getProperty(KafkaKey, kafkaKey, flowFile) || kafkaKey.empty()) {
Review comment:
if `MessageKeyField` worked before and now it no longer works we should remove it altogether and not mark it deprecated, if we would like to enable a "soft transition" we should mark it deprecated but further support the feature and produce loud warnings that it will not be so for long
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #1037: MINIFICPP-1520 Fix PublishKafka properties to support expression language
Posted by GitBox <gi...@apache.org>.
adamdebreceni commented on a change in pull request #1037:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1037#discussion_r599623112
##########
File path: extensions/librdkafka/PublishKafka.cpp
##########
@@ -925,12 +931,10 @@ void PublishKafka::onTrigger(const std::shared_ptr<core::ProcessContext> &contex
}
std::string kafkaKey;
- kafkaKey = "";
- if (context->getDynamicProperty(MessageKeyField, kafkaKey, flowFile) && !kafkaKey.empty()) {
- logger_->log_debug("PublishKafka: Message Key Field [%s]", kafkaKey);
- } else {
+ if (!context->getProperty(KafkaKey, kafkaKey, flowFile) || kafkaKey.empty()) {
Review comment:
your approach of keeping it as is, and logging errors if `MessageKeyField` is set, could be our best bet, given the constraints
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] fgerlits commented on a change in pull request #1037: MINIFICPP-1520 Fix PublishKafka properties to support expression language
Posted by GitBox <gi...@apache.org>.
fgerlits commented on a change in pull request #1037:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1037#discussion_r599586424
##########
File path: extensions/librdkafka/PublishKafka.cpp
##########
@@ -925,12 +931,10 @@ void PublishKafka::onTrigger(const std::shared_ptr<core::ProcessContext> &contex
}
std::string kafkaKey;
- kafkaKey = "";
- if (context->getDynamicProperty(MessageKeyField, kafkaKey, flowFile) && !kafkaKey.empty()) {
- logger_->log_debug("PublishKafka: Message Key Field [%s]", kafkaKey);
- } else {
+ if (!context->getProperty(KafkaKey, kafkaKey, flowFile) || kafkaKey.empty()) {
Review comment:
It's complicated. `MessageKeyField` used to "work" in the sense that if you set it (e.g. to `kafka.message.key` or `${kafka.message.key}`) then every message sent to Kafka would have the message key set to this literal value ("kafka.message.key" or "${kafka.message.key}") without any substitution, i.e. all messages would have the same key.
The description of the `MessageKeyField` property was "The name of a field in the Input Records that should be used as the Key for the Kafka message. Supports Expression Language: true (will be evaluated using flow file attributes)", every part of which is false.
I hope nobody was using this, but I think that silently changing the message keys to the flow file UUID (which is what will happen if you were using `MessageKeyField` previously and don't change anything) is better than keeping the old behavior where every message had the same key. Message keys are supposed to be unique.
What do you think of keeping the behavior as it is, but adding an error log if `MessageKeyField` is set?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #1037: MINIFICPP-1520 Fix PublishKafka properties to support expression language
Posted by GitBox <gi...@apache.org>.
adamdebreceni commented on a change in pull request #1037:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1037#discussion_r599599061
##########
File path: extensions/librdkafka/PublishKafka.cpp
##########
@@ -925,12 +931,10 @@ void PublishKafka::onTrigger(const std::shared_ptr<core::ProcessContext> &contex
}
std::string kafkaKey;
- kafkaKey = "";
- if (context->getDynamicProperty(MessageKeyField, kafkaKey, flowFile) && !kafkaKey.empty()) {
- logger_->log_debug("PublishKafka: Message Key Field [%s]", kafkaKey);
- } else {
+ if (!context->getProperty(KafkaKey, kafkaKey, flowFile) || kafkaKey.empty()) {
Review comment:
I don't think this ever worked since we try to get the `MessageKeyField` as if it were a dynamic property (`context->getDynamicProperty`) and since it is not a dynamic property we always did fall back to the flow file uuid
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #1037: MINIFICPP-1520 Fix PublishKafka properties to support expression language
Posted by GitBox <gi...@apache.org>.
adamdebreceni commented on a change in pull request #1037:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1037#discussion_r600432935
##########
File path: extensions/librdkafka/PublishKafka.cpp
##########
@@ -541,6 +543,11 @@ void PublishKafka::onSchedule(const std::shared_ptr<core::ProcessContext> &conte
conn_ = utils::make_unique<KafkaConnection>(key_);
configureNewConnection(context);
+ std::string message_key_field;
+ if (context->getProperty(MessageKeyField.getName(), message_key_field)) {
+ logger_->log_error("The %s property is set. This property is DEPRECATED and has no effect; please use Kafka Key instead.", MessageKeyField.getName());
Review comment:
I would go with error, as it uses a property that never worked and is only preserved to allow the processor to operate instead of fizzling out in onSchedule, personally I would even go for removing it, but I do not know how prevalent its usage is
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] fgerlits commented on a change in pull request #1037: MINIFICPP-1520 Fix PublishKafka properties to support expression language
Posted by GitBox <gi...@apache.org>.
fgerlits commented on a change in pull request #1037:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1037#discussion_r600635867
##########
File path: PROCESSORS.md
##########
@@ -971,13 +971,13 @@ In the list below, the names of required properties appear in bold. Any other pr
|**Known Brokers**|||A comma-separated list of known Kafka Brokers in the format <host>:<port><br/>**Supports Expression Language: true**|
|Max Flow Segment Size|0 B||Maximum flow content payload segment size for the kafka record. 0 B means unlimited.|
|Max Request Size|||Maximum Kafka protocol request message size|
-|Message Key Field|||The name of a field in the Input Records that should be used as the Key for the Kafka message.
-Supports Expression Language: true (will be evaluated using flow file attributes)|
-|Message Timeout|30 sec||The total time sending a message could take<br/>**Supports Expression Language: true**|
+|Kafka Key|||The key to use for the message. If not specified, the UUID of the flow file is used as the message key.<br/>**Supports Expression Language: true**|
Review comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni closed pull request #1037: MINIFICPP-1520 Fix PublishKafka properties to support expression language
Posted by GitBox <gi...@apache.org>.
adamdebreceni closed pull request #1037:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1037
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] szaszm commented on a change in pull request #1037: MINIFICPP-1520 Fix PublishKafka properties to support expression language
Posted by GitBox <gi...@apache.org>.
szaszm commented on a change in pull request #1037:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1037#discussion_r600491028
##########
File path: extensions/librdkafka/PublishKafka.cpp
##########
@@ -541,6 +543,11 @@ void PublishKafka::onSchedule(const std::shared_ptr<core::ProcessContext> &conte
conn_ = utils::make_unique<KafkaConnection>(key_);
configureNewConnection(context);
+ std::string message_key_field;
+ if (context->getProperty(MessageKeyField.getName(), message_key_field)) {
+ logger_->log_error("The %s property is set. This property is DEPRECATED and has no effect; please use Kafka Key instead.", MessageKeyField.getName());
Review comment:
Having the property set is likely a mistake, but it doesn't cause data loss or failures down the line. Seeing this message doesn't mean a change in behavior from the happy path, only that the flow is incorrect, but still runnable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #1037: MINIFICPP-1520 Fix PublishKafka properties to support expression language
Posted by GitBox <gi...@apache.org>.
adamdebreceni commented on a change in pull request #1037:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1037#discussion_r599599061
##########
File path: extensions/librdkafka/PublishKafka.cpp
##########
@@ -925,12 +931,10 @@ void PublishKafka::onTrigger(const std::shared_ptr<core::ProcessContext> &contex
}
std::string kafkaKey;
- kafkaKey = "";
- if (context->getDynamicProperty(MessageKeyField, kafkaKey, flowFile) && !kafkaKey.empty()) {
- logger_->log_debug("PublishKafka: Message Key Field [%s]", kafkaKey);
- } else {
+ if (!context->getProperty(KafkaKey, kafkaKey, flowFile) || kafkaKey.empty()) {
Review comment:
I don't think this ever worked since we try to get the `MessageKeyField` as if it were a dynamic property (`context->getDynamicProperty`)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] szaszm commented on a change in pull request #1037: MINIFICPP-1520 Fix PublishKafka properties to support expression language
Posted by GitBox <gi...@apache.org>.
szaszm commented on a change in pull request #1037:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1037#discussion_r600426575
##########
File path: extensions/librdkafka/PublishKafka.cpp
##########
@@ -541,6 +543,11 @@ void PublishKafka::onSchedule(const std::shared_ptr<core::ProcessContext> &conte
conn_ = utils::make_unique<KafkaConnection>(key_);
configureNewConnection(context);
+ std::string message_key_field;
+ if (context->getProperty(MessageKeyField.getName(), message_key_field)) {
+ logger_->log_error("The %s property is set. This property is DEPRECATED and has no effect; please use Kafka Key instead.", MessageKeyField.getName());
Review comment:
I would prefer this to be a warning, not an error
##########
File path: extensions/librdkafka/tests/CMakeLists.txt
##########
@@ -29,10 +29,12 @@ FOREACH(testfile ${KAFKA_TESTS})
createTests("${testfilename}")
MATH(EXPR KAFKA_TEST_COUNT "${KAFKA_TEST_COUNT}+1")
# The line below handles integration test
+ target_include_directories(${testfilename} BEFORE PRIVATE "../..")
Review comment:
The comment above is outdated, I suggest removing it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] fgerlits commented on a change in pull request #1037: MINIFICPP-1520 Fix PublishKafka properties to support expression language
Posted by GitBox <gi...@apache.org>.
fgerlits commented on a change in pull request #1037:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1037#discussion_r600522506
##########
File path: extensions/librdkafka/PublishKafka.cpp
##########
@@ -541,6 +543,11 @@ void PublishKafka::onSchedule(const std::shared_ptr<core::ProcessContext> &conte
conn_ = utils::make_unique<KafkaConnection>(key_);
configureNewConnection(context);
+ std::string message_key_field;
+ if (context->getProperty(MessageKeyField.getName(), message_key_field)) {
+ logger_->log_error("The %s property is set. This property is DEPRECATED and has no effect; please use Kafka Key instead.", MessageKeyField.getName());
Review comment:
I'm fine with logging it on either warning or error level, so we'll need a 4th vote to decide. :)
My reasoning for an error log was that if the user has set the MessageKeyField property, then their flow is not working as intended. E.g. they may have wanted to set the message key to the value of a particular flow file attribute, but it is set to the flow file UUID instead -- this can cause difficult-to-debug problems down the line.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] fgerlits commented on a change in pull request #1037: MINIFICPP-1520 Fix PublishKafka properties to support expression language
Posted by GitBox <gi...@apache.org>.
fgerlits commented on a change in pull request #1037:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1037#discussion_r599740665
##########
File path: extensions/librdkafka/PublishKafka.cpp
##########
@@ -925,12 +931,10 @@ void PublishKafka::onTrigger(const std::shared_ptr<core::ProcessContext> &contex
}
std::string kafkaKey;
- kafkaKey = "";
- if (context->getDynamicProperty(MessageKeyField, kafkaKey, flowFile) && !kafkaKey.empty()) {
- logger_->log_debug("PublishKafka: Message Key Field [%s]", kafkaKey);
- } else {
+ if (!context->getProperty(KafkaKey, kafkaKey, flowFile) || kafkaKey.empty()) {
Review comment:
You're right about both MiNiFi startup and the old (non-)behavior or `MessageKeyField`.
I have added an error log in `onSchedule()` when `MessageKeyField` is set.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #1037: MINIFICPP-1520 Fix PublishKafka properties to support expression language
Posted by GitBox <gi...@apache.org>.
adamdebreceni commented on a change in pull request #1037:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1037#discussion_r600565978
##########
File path: PROCESSORS.md
##########
@@ -971,13 +971,13 @@ In the list below, the names of required properties appear in bold. Any other pr
|**Known Brokers**|||A comma-separated list of known Kafka Brokers in the format <host>:<port><br/>**Supports Expression Language: true**|
|Max Flow Segment Size|0 B||Maximum flow content payload segment size for the kafka record. 0 B means unlimited.|
|Max Request Size|||Maximum Kafka protocol request message size|
-|Message Key Field|||The name of a field in the Input Records that should be used as the Key for the Kafka message.
-Supports Expression Language: true (will be evaluated using flow file attributes)|
-|Message Timeout|30 sec||The total time sending a message could take<br/>**Supports Expression Language: true**|
+|Kafka Key|||The key to use for the message. If not specified, the UUID of the flow file is used as the message key.<br/>**Supports Expression Language: true**|
Review comment:
nitpick: other properties seem to be in an alphabetical order
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] fgerlits commented on a change in pull request #1037: MINIFICPP-1520 Fix PublishKafka properties to support expression language
Posted by GitBox <gi...@apache.org>.
fgerlits commented on a change in pull request #1037:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1037#discussion_r599740665
##########
File path: extensions/librdkafka/PublishKafka.cpp
##########
@@ -925,12 +931,10 @@ void PublishKafka::onTrigger(const std::shared_ptr<core::ProcessContext> &contex
}
std::string kafkaKey;
- kafkaKey = "";
- if (context->getDynamicProperty(MessageKeyField, kafkaKey, flowFile) && !kafkaKey.empty()) {
- logger_->log_debug("PublishKafka: Message Key Field [%s]", kafkaKey);
- } else {
+ if (!context->getProperty(KafkaKey, kafkaKey, flowFile) || kafkaKey.empty()) {
Review comment:
You're right about both MiNiFi startup and the old (non-)behavior of `MessageKeyField`.
I have added an error log in `onSchedule()` when `MessageKeyField` is set.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #1037: MINIFICPP-1520 Fix PublishKafka properties to support expression language
Posted by GitBox <gi...@apache.org>.
adamdebreceni commented on a change in pull request #1037:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1037#discussion_r599602610
##########
File path: extensions/librdkafka/PublishKafka.cpp
##########
@@ -925,12 +931,10 @@ void PublishKafka::onTrigger(const std::shared_ptr<core::ProcessContext> &contex
}
std::string kafkaKey;
- kafkaKey = "";
- if (context->getDynamicProperty(MessageKeyField, kafkaKey, flowFile) && !kafkaKey.empty()) {
- logger_->log_debug("PublishKafka: Message Key Field [%s]", kafkaKey);
- } else {
+ if (!context->getProperty(KafkaKey, kafkaKey, flowFile) || kafkaKey.empty()) {
Review comment:
does an exception in an onSchedule really prevents a MiNiFi startup? that would be really bad, as there is no way to fix such a malformed flow through a c2 server
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] fgerlits commented on a change in pull request #1037: MINIFICPP-1520 Fix PublishKafka properties to support expression language
Posted by GitBox <gi...@apache.org>.
fgerlits commented on a change in pull request #1037:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1037#discussion_r599586424
##########
File path: extensions/librdkafka/PublishKafka.cpp
##########
@@ -925,12 +931,10 @@ void PublishKafka::onTrigger(const std::shared_ptr<core::ProcessContext> &contex
}
std::string kafkaKey;
- kafkaKey = "";
- if (context->getDynamicProperty(MessageKeyField, kafkaKey, flowFile) && !kafkaKey.empty()) {
- logger_->log_debug("PublishKafka: Message Key Field [%s]", kafkaKey);
- } else {
+ if (!context->getProperty(KafkaKey, kafkaKey, flowFile) || kafkaKey.empty()) {
Review comment:
It's complicated. `MessageKeyField` used to "work" in the sense that if you set it (e.g. to `kafka.message.key` or `${kafka.message.key}`) then every message sent to Kafka would have the message key set to this literal value ("kafka.message.key" or "${kafka.message.key}") without any substitution, i.e. all messages would have the same key.
The description of the `MessageKeyField` property was "The name of a field in the Input Records that should be used as the Key for the Kafka message. Supports Expression Language: true (will be evaluated using flow file attributes)", every part of which is false.
I hope nobody was using this, but I think that silently changing the message keys to the flow file UUID (which is what will happen if you were using `MessageKeyField` previously and don't change anything) is better than keeping the old behavior where every message had the same key. Message keys are supposed to be unique.
We can't simply remove `MessageKeyField`, because then it will be treated as a dynamic property, and dynamic properties are required to be valid Kafka configuration property names. So if we remove the `MessageKeyField` property and it is present in the user's `config.yml` file, then `PublishKafka::onSchedule` will throw, and MiNiFi will not start up.
What do you think of keeping the behavior as it is, but adding an error log if `MessageKeyField` is set?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi-minifi-cpp] adamdebreceni commented on a change in pull request #1037: MINIFICPP-1520 Fix PublishKafka properties to support expression language
Posted by GitBox <gi...@apache.org>.
adamdebreceni commented on a change in pull request #1037:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1037#discussion_r600565978
##########
File path: PROCESSORS.md
##########
@@ -971,13 +971,13 @@ In the list below, the names of required properties appear in bold. Any other pr
|**Known Brokers**|||A comma-separated list of known Kafka Brokers in the format <host>:<port><br/>**Supports Expression Language: true**|
|Max Flow Segment Size|0 B||Maximum flow content payload segment size for the kafka record. 0 B means unlimited.|
|Max Request Size|||Maximum Kafka protocol request message size|
-|Message Key Field|||The name of a field in the Input Records that should be used as the Key for the Kafka message.
-Supports Expression Language: true (will be evaluated using flow file attributes)|
-|Message Timeout|30 sec||The total time sending a message could take<br/>**Supports Expression Language: true**|
+|Kafka Key|||The key to use for the message. If not specified, the UUID of the flow file is used as the message key.<br/>**Supports Expression Language: true**|
Review comment:
nitpick: other properties seem to be in alphabetical order
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org