review changes

Author: martinzink

extensions/kafka/ConsumeKafka.cpp

@@ -114,7 +114,7 @@ void ConsumeKafka::configureNewConnection(api::core::ProcessContext& context) {
   // Registering a rebalance_cb turns off librdkafka's automatic partition assignment/revocation and instead delegates that
   // responsibility to the application's rebalance_cb.
   if (commit_policy_ != consume_kafka::CommitPolicyEnum::CommitFromIncomingFlowFiles) {
-    rd_kafka_conf_set_rebalance_cb(conf_.get(), utils::KafkaOpaque::rebalance_cb);
+    rd_kafka_conf_set_rebalance_cb(conf_.get(), utils::KafkaOpaque::rebalanceCallback);
   }
 
   setKafkaAuthenticationParameters(context, gsl::make_not_null(conf_.get()));

extensions/kafka/PublishKafka.cpp

@@ -595,7 +595,7 @@ bool PublishKafka::createNewTopic(const api::core::ProcessContext& context, cons
 }
 
 std::optional<api::utils::net::SslData> PublishKafka::getSslData(api::core::ProcessContext& context) const {
-  if (auto result = KafkaProcessorBase::getSslData(context); result) { return result; }
+  if (auto result = context.getSslData(SSLContextService); result) { return *result; }
 
   api::utils::net::SslData ssl_data;
   if (auto security_ca = context.getProperty(SecurityCA, nullptr)) { ssl_data.ca_loc = *security_ca; }

extensions/kafka/rdkafka_utils.cpp

@@ -36,7 +36,7 @@ void setKafkaConfigurationField(rd_kafka_conf_t& configuration, const std::strin
   }
 }
 
-void KafkaOpaque::print_topics_list(const rd_kafka_topic_partition_list_t& kf_topic_partition_list) const {
+void KafkaOpaque::printTopicsList(const rd_kafka_topic_partition_list_t& kf_topic_partition_list) const {
   if (!logger_.should_log(core::logging::debug))
     return;
   for (int i = 0; i < kf_topic_partition_list.cnt; ++i) {
@@ -71,7 +71,7 @@ void KafkaOpaque::logCallback(const rd_kafka_t* rk, const int level, const char*
     default: gsl_FailFast();
   }
 }
-void KafkaOpaque::rebalance_cb(rd_kafka_t* rk, const rd_kafka_resp_err_t trigger, rd_kafka_topic_partition_list_t* partitions, void* opaque_ptr) {
+void KafkaOpaque::rebalanceCallback(rd_kafka_t* rk, const rd_kafka_resp_err_t trigger, rd_kafka_topic_partition_list_t* partitions, void* opaque_ptr) {
   const auto* kafka_opaque = static_cast<KafkaOpaque*>(opaque_ptr);
   if (!kafka_opaque) {
     return;
@@ -81,14 +81,14 @@ void KafkaOpaque::rebalance_cb(rd_kafka_t* rk, const rd_kafka_resp_err_t trigger
   switch (trigger) {
     case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS:
       kafka_opaque->logger_.log_debug("assigned:");
-      kafka_opaque->print_topics_list(*partitions);
+      kafka_opaque->printTopicsList(*partitions);
       assign_error = rd_kafka_assign(rk, partitions);
       break;
 
     case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS:
       kafka_opaque->logger_.log_debug("revoked:");
       rd_kafka_commit(rk, partitions, /* async = */ 0);  // Sync commit, maybe unnecessary
-      kafka_opaque->print_topics_list(*partitions);
+      kafka_opaque->printTopicsList(*partitions);
 
       assign_error = rd_kafka_assign(rk, nullptr);
       break;

extensions/kafka/rdkafka_utils.h

@@ -100,9 +100,9 @@ class KafkaOpaque {
  public:
   explicit KafkaOpaque(core::logging::Logger& logger) : logger_(logger) {}
 
-  void print_topics_list(const rd_kafka_topic_partition_list_t& kf_topic_partition_list) const;
+  void printTopicsList(const rd_kafka_topic_partition_list_t& kf_topic_partition_list) const;
   static void logCallback(const rd_kafka_t* rk, int level, const char* /*fac*/, const char* buf);
-  static void rebalance_cb(rd_kafka_t* rk, rd_kafka_resp_err_t trigger, rd_kafka_topic_partition_list_t* partitions, void* opaque_ptr);
+  static void rebalanceCallback(rd_kafka_t* rk, rd_kafka_resp_err_t trigger, rd_kafka_topic_partition_list_t* partitions, void* opaque_ptr);
 
  private:
   core::logging::Logger& logger_;