commit c759c96c9ab6f59a98a624dd99a9540d73d8d6b4
author Ferenc Gerlits <fgerlits@gmail.com> Fri Apr 09 18:04:47 2021 +0200
committer Arpad Boda <aboda@apache.org> Tue Apr 20 17:34:40 2021 +0200
tree 706fa059e5d0e548c7295f647f7321a239356ac8
parent a29b414b8f7f5310c88a51e9448ec8ce3a83b4c0

MINIFICPP-1539 Check if Message Key Field is really set

Signed-off-by: Arpad Boda <aboda@apache.org>

This closes #1049

extensions/librdkafka/PublishKafka.cpp

diff --git a/extensions/librdkafka/PublishKafka.cpp b/extensions/librdkafka/PublishKafka.cpp
index ceb76f3..b83d028 100644
--- a/extensions/librdkafka/PublishKafka.cpp
+++ b/extensions/librdkafka/PublishKafka.cpp
@@ -544,7 +544,7 @@
   configureNewConnection(context);
 
   std::string message_key_field;
-  if (context->getProperty(MessageKeyField.getName(), message_key_field)) {
+  if (context->getProperty(MessageKeyField.getName(), message_key_field) && !message_key_field.empty()) {
     logger_->log_error("The %s property is set. This property is DEPRECATED and has no effect; please use Kafka Key instead.", MessageKeyField.getName());
   }
 

extensions/librdkafka/tests/PublishKafkaTests.cpp

diff --git a/extensions/librdkafka/tests/PublishKafkaTests.cpp b/extensions/librdkafka/tests/PublishKafkaTests.cpp
index ea1c4ef..db44eb3 100644
--- a/extensions/librdkafka/tests/PublishKafkaTests.cpp
+++ b/extensions/librdkafka/tests/PublishKafkaTests.cpp
@@ -190,3 +190,19 @@
 
   REQUIRE(LogTestController::getInstance().contains("PublishKafka: DynamicProperty: [retry.backoff.ms] -> [180000]"));
 }
+
+TEST_CASE("PublishKafka complains if Message Key Field is set, but only if it is set", "[PublishKafka][properties]") {
+  PublishKafkaTestRunner test_runner;
+
+  SECTION("Message Key Field is not set, so there is no error") {
+    test_runner.runProcessor();
+    REQUIRE_FALSE(LogTestController::getInstance().contains("error"));
+  }
+
+  SECTION("Message Key Field is set, so there is an error log") {
+    test_runner.setPublishKafkaProperty(processors::PublishKafka::MessageKeyField, "kafka.key");
+    test_runner.runProcessor();
+    REQUIRE(LogTestController::getInstance().contains("The " + processors::PublishKafka::MessageKeyField.getName() +
+        " property is set. This property is DEPRECATED and has no effect"));
+  }
+}