diff --git a/Windows.md b/Windows.md
index d453955..c2400be 100644
--- a/Windows.md
+++ b/Windows.md
@@ -62,6 +62,7 @@
 | /J | Enables JNI |
 | /64 | Creates 64-bit build instead of a 32-bit one |
 | /D | Builds RelWithDebInfo build instead of Release |
+| /DD | Builds Debug build instead of Release |
 | /CI | Sets STRICT_GSL_CHECKS to AUDIT |
 
 Examples:
diff --git a/extensions/librdkafka/KafkaConnection.cpp b/extensions/librdkafka/KafkaConnection.cpp
index cf871a8..3d412ad 100644
--- a/extensions/librdkafka/KafkaConnection.cpp
+++ b/extensions/librdkafka/KafkaConnection.cpp
@@ -87,7 +87,7 @@
   return nullptr;
 }
 
-KafkaConnectionKey const * const KafkaConnection::getKey() const {
+KafkaConnectionKey const* KafkaConnection::getKey() const {
   return &key_;
 }
 
diff --git a/extensions/librdkafka/KafkaConnection.h b/extensions/librdkafka/KafkaConnection.h
index 816b40a..8d5b12e 100644
--- a/extensions/librdkafka/KafkaConnection.h
+++ b/extensions/librdkafka/KafkaConnection.h
@@ -37,14 +37,16 @@
 namespace minifi {
 namespace processors {
 
-class KafkaConnectionKey {
- public:
-    std::string client_id_;
-    std::string brokers_;
+struct KafkaConnectionKey {
+  std::string client_id_;
+  std::string brokers_;
 
-    bool operator <(const KafkaConnectionKey& rhs) const {
-      return std::tie(brokers_, client_id_) < std::tie(rhs.brokers_, rhs.client_id_);
-    }
+  bool operator< (const KafkaConnectionKey& rhs) const { return std::tie(brokers_, client_id_) <  std::tie(rhs.brokers_, rhs.client_id_); }
+  bool operator<=(const KafkaConnectionKey& rhs) const { return std::tie(brokers_, client_id_) <= std::tie(rhs.brokers_, rhs.client_id_); }
+  bool operator==(const KafkaConnectionKey& rhs) const { return std::tie(brokers_, client_id_) == std::tie(rhs.brokers_, rhs.client_id_); }
+  bool operator!=(const KafkaConnectionKey& rhs) const { return std::tie(brokers_, client_id_) != std::tie(rhs.brokers_, rhs.client_id_); }
+  bool operator> (const KafkaConnectionKey& rhs) const { return std::tie(brokers_, client_id_) >  std::tie(rhs.brokers_, rhs.client_id_); }
+  bool operator>=(const KafkaConnectionKey& rhs) const { return std::tie(brokers_, client_id_) >= std::tie(rhs.brokers_, rhs.client_id_); }
 };
 
 class KafkaConnection {
@@ -70,7 +72,7 @@
 
   std::shared_ptr<KafkaTopic> getTopic(const std::string &topic) const;
 
-  KafkaConnectionKey const * const getKey() const;
+  KafkaConnectionKey const* getKey() const;
 
   void putTopic(const std::string &topicName, const std::shared_ptr<KafkaTopic> &topic);
 
diff --git a/extensions/librdkafka/PublishKafka.cpp b/extensions/librdkafka/PublishKafka.cpp
index fc5ee56..1b92edc 100644
--- a/extensions/librdkafka/PublishKafka.cpp
+++ b/extensions/librdkafka/PublishKafka.cpp
@@ -39,6 +39,21 @@
 namespace minifi {
 namespace processors {
 
+#define COMPRESSION_CODEC_NONE "none"
+#define COMPRESSION_CODEC_GZIP "gzip"
+#define COMPRESSION_CODEC_SNAPPY "snappy"
+#define ROUND_ROBIN_PARTITIONING "Round Robin"
+#define RANDOM_PARTITIONING "Random Robin"
+#define USER_DEFINED_PARTITIONING "User-Defined"
+#define DELIVERY_REPLICATED "all"
+#define DELIVERY_ONE_NODE "1"
+#define DELIVERY_BEST_EFFORT "0"
+#define SECURITY_PROTOCOL_PLAINTEXT "plaintext"
+#define SECURITY_PROTOCOL_SSL "ssl"
+#define SECURITY_PROTOCOL_SASL_PLAINTEXT "sasl_plaintext"
+#define SECURITY_PROTOCOL_SASL_SSL "sasl_ssl"
+#define KAFKA_KEY_ATTRIBUTE "kafka.key"
+
 const core::Property PublishKafka::SeedBrokers(
     core::PropertyBuilder::createProperty("Known Brokers")->withDescription("A comma-separated list of known Kafka Brokers in the format <host>:<port>")
         ->isRequired(true)->supportsExpressionLanguage(true)->build());
@@ -144,6 +159,305 @@
 struct rd_kafka_topic_conf_deleter {
   void operator()(rd_kafka_topic_conf_t* p) const noexcept { rd_kafka_topic_conf_destroy(p); }
 };
+
+// Message
+enum class MessageStatus : uint8_t {
+  InFlight,
+  Error,
+  Success
+};
+
+const char* to_string(const MessageStatus s) {
+  switch (s) {
+    case MessageStatus::InFlight: return "InFlight";
+    case MessageStatus::Error: return "Error";
+    case MessageStatus::Success: return "Success";
+  }
+  throw std::runtime_error{"PublishKafka to_string(MessageStatus): unreachable code"};
+}
+
+struct MessageResult {
+  MessageStatus status = MessageStatus::InFlight;
+  rd_kafka_resp_err_t err_code = RD_KAFKA_RESP_ERR_NO_ERROR;
+};
+
+struct FlowFileResult {
+  bool flow_file_error = false;
+  std::vector<MessageResult> messages;
+};
+}  // namespace
+
+class PublishKafka::Messages {
+  std::mutex mutex_;
+  std::condition_variable cv_;
+  std::vector<FlowFileResult> flow_files_;
+  bool interrupted_ = false;
+  const std::shared_ptr<logging::Logger> logger_;
+
+  std::string logStatus(const std::unique_lock<std::mutex>& lock) const {
+    gsl_Expects(lock.owns_lock());
+    const auto messageresult_ok = [](const MessageResult r) { return r.status == MessageStatus::Success && r.err_code == RD_KAFKA_RESP_ERR_NO_ERROR; };
+    const auto messageresult_inflight = [](const MessageResult r) { return r.status == MessageStatus::InFlight && r.err_code == RD_KAFKA_RESP_ERR_NO_ERROR; };
+    std::vector<size_t> flow_files_in_flight;
+    std::ostringstream oss;
+    if (interrupted_) { oss << "interrupted, "; }
+    for (size_t ffi = 0; ffi < flow_files_.size(); ++ffi) {
+      const auto& flow_file = flow_files_[ffi];
+      if (!flow_file.flow_file_error && std::all_of(std::begin(flow_file.messages), std::end(flow_file.messages), messageresult_ok)) {
+        continue;  // don't log the happy path to reduce log spam
+      }
+      if (!flow_file.flow_file_error && std::all_of(std::begin(flow_file.messages), std::end(flow_file.messages), messageresult_inflight)) {
+        flow_files_in_flight.push_back(ffi);
+        continue;  // don't log fully in-flight flow files here, log them at the end instead
+      }
+      oss << '[' << ffi << "]: {";
+      if (flow_file.flow_file_error) { oss << "error, "; }
+      for (size_t msgi = 0; msgi < flow_file.messages.size(); ++msgi) {
+        const auto& msg = flow_file.messages[msgi];
+        if (messageresult_ok(msg)) {
+          continue;
+        }
+        oss << '<' << msgi << ">: (msg " << to_string(msg.status) << ", " << rd_kafka_err2str(msg.err_code) << "), ";
+      }
+      oss << "}, ";
+    }
+    oss << "in-flight (" << flow_files_in_flight.size() << "): " << utils::StringUtils::join(",", flow_files_in_flight);
+    return oss.str();
+  }
+
+ public:
+  explicit Messages(std::shared_ptr<logging::Logger> logger) :logger_{std::move(logger)} {}
+
+  void waitForCompletion() {
+    std::unique_lock<std::mutex> lock(mutex_);
+    cv_.wait(lock, [this, &lock] {
+      if (logger_->should_log(logging::LOG_LEVEL::trace)) {
+        logger_->log_trace("%s", logStatus(lock));
+      }
+      return interrupted_ || std::all_of(std::begin(this->flow_files_), std::end(this->flow_files_), [](const FlowFileResult& flow_file) {
+        return flow_file.flow_file_error || std::all_of(std::begin(flow_file.messages), std::end(flow_file.messages), [](const MessageResult& message) {
+          return message.status != MessageStatus::InFlight;
+        });
+      });
+    });
+  }
+
+  template<typename Func>
+  auto modifyResult(size_t index, Func fun) -> decltype(fun(flow_files_.at(index))) {
+    std::unique_lock<std::mutex> lock(mutex_);
+    const auto notifier = gsl::finally([this]{ cv_.notify_all(); });
+    try {
+      return fun(flow_files_.at(index));
+    } catch(const std::exception& ex) {
+      logger_->log_warn("Messages::modifyResult exception: %s", ex.what());
+      throw;
+    }
+  }
+
+  size_t addFlowFile() {
+    std::lock_guard<std::mutex> lock(mutex_);
+    flow_files_.emplace_back();
+    return flow_files_.size() - 1;
+  }
+
+  template<typename Func>
+  auto iterateFlowFiles(Func fun) -> utils::void_t<decltype(fun(size_t{0}, flow_files_.front()))> {
+    std::lock_guard<std::mutex> lock(mutex_);
+    for (size_t index = 0U; index < flow_files_.size(); index++) {
+      fun(index, flow_files_[index]);
+    }
+  }
+
+  void interrupt() {
+    std::unique_lock<std::mutex> lock(mutex_);
+    interrupted_ = true;
+    cv_.notify_all();
+    gsl_Ensures(interrupted_);
+  }
+
+  bool wasInterrupted() {
+    std::lock_guard<std::mutex> lock(mutex_);
+    return interrupted_;
+  }
+};
+
+namespace {
+class ReadCallback : public InputStreamCallback {
+ public:
+  struct rd_kafka_headers_deleter {
+    void operator()(rd_kafka_headers_t* ptr) const noexcept {
+      rd_kafka_headers_destroy(ptr);
+    }
+  };
+
+  using rd_kafka_headers_unique_ptr = std::unique_ptr<rd_kafka_headers_t, rd_kafka_headers_deleter>;
+
+ private:
+  void allocate_message_object(const size_t segment_num) const {
+    messages_->modifyResult(flow_file_index_, [segment_num](FlowFileResult& flow_file) {
+      // allocate message object to be filled in by the callback in produce()
+      if (flow_file.messages.size() < segment_num + 1) {
+        flow_file.messages.resize(segment_num + 1);
+      }
+      gsl_Ensures(flow_file.messages.size() > segment_num);
+    });
+  }
+
+  static rd_kafka_headers_unique_ptr make_headers(const core::FlowFile& flow_file, utils::Regex& attribute_name_regex) {
+    const gsl::owner<rd_kafka_headers_t*> result{ rd_kafka_headers_new(8) };
+    if (!result) { throw std::bad_alloc{}; }
+
+    for (const auto& kv : flow_file.getAttributes()) {
+      if (attribute_name_regex.match(kv.first)) {
+        rd_kafka_header_add(result, kv.first.c_str(), kv.first.size(), kv.second.c_str(), kv.second.size());
+      }
+    }
+    return rd_kafka_headers_unique_ptr{ result };
+  }
+
+  rd_kafka_resp_err_t produce(const size_t segment_num, std::vector<unsigned char>& buffer, const size_t buflen) const {
+    const std::shared_ptr<PublishKafka::Messages> messages_ptr_copy = this->messages_;
+    const auto flow_file_index_copy = this->flow_file_index_;
+    const auto logger = logger_;
+    const auto produce_callback = [messages_ptr_copy, flow_file_index_copy, segment_num, logger](rd_kafka_t * /*rk*/, const rd_kafka_message_t *rkmessage) {
+      messages_ptr_copy->modifyResult(flow_file_index_copy, [segment_num, rkmessage, logger, flow_file_index_copy](FlowFileResult &flow_file) {
+        auto &message = flow_file.messages.at(segment_num);
+        message.err_code = rkmessage->err;
+        message.status = message.err_code == 0 ? MessageStatus::Success : MessageStatus::Error;
+        if (message.err_code != RD_KAFKA_RESP_ERR_NO_ERROR) {
+          logger->log_warn("delivery callback, flow file #%zu/segment #%zu: %s", flow_file_index_copy, segment_num, rd_kafka_err2str(message.err_code));
+        } else {
+          logger->log_debug("delivery callback, flow file #%zu/segment #%zu: success", flow_file_index_copy, segment_num);
+        }
+      });
+    };
+    // release()d below, deallocated in PublishKafka::messageDeliveryCallback
+    auto callback_ptr = utils::make_unique<std::function<void(rd_kafka_t*, const rd_kafka_message_t*)>>(std::move(produce_callback));
+
+    allocate_message_object(segment_num);
+
+    const gsl::owner<rd_kafka_headers_t*> hdrs_copy = rd_kafka_headers_copy(hdrs.get());
+    const auto err = rd_kafka_producev(rk_, RD_KAFKA_V_RKT(rkt_), RD_KAFKA_V_PARTITION(RD_KAFKA_PARTITION_UA), RD_KAFKA_V_MSGFLAGS(RD_KAFKA_MSG_F_COPY), RD_KAFKA_V_VALUE(buffer.data(), buflen),
+        RD_KAFKA_V_HEADERS(hdrs_copy), RD_KAFKA_V_KEY(key_.c_str(), key_.size()), RD_KAFKA_V_OPAQUE(callback_ptr.get()), RD_KAFKA_V_END);
+    if (err == RD_KAFKA_RESP_ERR_NO_ERROR) {
+      // in case of failure, messageDeliveryCallback is not called and callback_ptr will delete the callback
+      // in case of success, messageDeliveryCallback takes ownership of the callback, so we no longer need to delete it
+      (void)callback_ptr.release();
+    } else {
+      // in case of failure, rd_kafka_producev doesn't take ownership of the headers, so we need to delete them
+      rd_kafka_headers_destroy(hdrs_copy);
+    }
+    logger_->log_trace("produce enqueued flow file #%zu/segment #%zu: %s", flow_file_index_, segment_num, rd_kafka_err2str(err));
+    return err;
+  }
+
+ public:
+  ReadCallback(const uint64_t max_seg_size,
+      std::string key,
+      rd_kafka_topic_t* const rkt,
+      rd_kafka_t* const rk,
+      const core::FlowFile& flowFile,
+      utils::Regex& attributeNameRegex,
+      std::shared_ptr<PublishKafka::Messages> messages,
+      const size_t flow_file_index,
+      const bool fail_empty_flow_files,
+      std::shared_ptr<logging::Logger> logger)
+      : flow_size_(flowFile.getSize()),
+      max_seg_size_(max_seg_size == 0 || flow_size_ < max_seg_size ? flow_size_ : max_seg_size),
+      key_(std::move(key)),
+      rkt_(rkt),
+      rk_(rk),
+      hdrs(make_headers(flowFile, attributeNameRegex)),
+      messages_(std::move(messages)),
+      flow_file_index_(flow_file_index),
+      fail_empty_flow_files_(fail_empty_flow_files),
+      logger_(std::move(logger))
+  { }
+
+  ReadCallback(const ReadCallback&) = delete;
+  ReadCallback& operator=(ReadCallback) = delete;
+
+  int64_t process(const std::shared_ptr<io::BaseStream> stream) override {
+    std::vector<unsigned char> buffer;
+
+    buffer.resize(max_seg_size_);
+    read_size_ = 0;
+    status_ = 0;
+    called_ = true;
+
+    gsl_Expects(max_seg_size_ != 0 || flow_size_ == 0 && "max_seg_size_ == 0 implies flow_size_ == 0");
+    // ^^ therefore checking max_seg_size_ == 0 handles both division by zero and flow_size_ == 0 cases
+    const size_t reserved_msg_capacity = max_seg_size_ == 0 ? 1 : utils::intdiv_ceil(flow_size_, max_seg_size_);
+    messages_->modifyResult(flow_file_index_, [reserved_msg_capacity](FlowFileResult& flow_file) {
+      flow_file.messages.reserve(reserved_msg_capacity);
+    });
+
+    // If the flow file is empty, we still want to send the message, unless the user wants to fail_empty_flow_files_
+    if (flow_size_ == 0 && !fail_empty_flow_files_) {
+      const auto err = produce(0, buffer, 0);
+      if (err != RD_KAFKA_RESP_ERR_NO_ERROR) {
+        status_ = -1;
+        error_ = rd_kafka_err2str(err);
+      }
+      return 0;
+    }
+
+    for (size_t segment_num = 0; read_size_ < flow_size_; ++segment_num) {
+      const int readRet = stream->read(buffer.data(), buffer.size());
+      if (readRet < 0) {
+        status_ = -1;
+        error_ = "Failed to read from stream";
+        return read_size_;
+      }
+
+      if (readRet <= 0) { break; }
+
+      const auto err = produce(segment_num, buffer, readRet);
+      if (err) {
+        messages_->modifyResult(flow_file_index_, [segment_num, err](FlowFileResult& flow_file) {
+          auto& message = flow_file.messages.at(segment_num);
+          message.status = MessageStatus::Error;
+          message.err_code = err;
+        });
+        status_ = -1;
+        error_ = rd_kafka_err2str(err);
+        return read_size_;
+      }
+      read_size_ += readRet;
+    }
+    return read_size_;
+  }
+
+  const uint64_t flow_size_ = 0;
+  const uint64_t max_seg_size_ = 0;
+  const std::string key_;
+  rd_kafka_topic_t* const rkt_ = nullptr;
+  rd_kafka_t* const rk_ = nullptr;
+  const rd_kafka_headers_unique_ptr hdrs;  // not null
+  const std::shared_ptr<PublishKafka::Messages> messages_;
+  const size_t flow_file_index_;
+  int status_ = 0;
+  std::string error_;
+  int read_size_ = 0;
+  bool called_ = false;
+  const bool fail_empty_flow_files_ = true;
+  const std::shared_ptr<logging::Logger> logger_;
+};
+
+/**
+ * Message delivery report callback using the richer rd_kafka_message_t object.
+ */
+void messageDeliveryCallback(rd_kafka_t* rk, const rd_kafka_message_t* rkmessage, void* /*opaque*/) {
+  if (rkmessage->_private == nullptr) {
+    return;
+  }
+  // allocated in ReadCallback::produce
+  auto* const func = reinterpret_cast<std::function<void(rd_kafka_t*, const rd_kafka_message_t*)>*>(rkmessage->_private);
+  try {
+    (*func)(rk, rkmessage);
+  } catch (...) { }
+  delete func;
+}
 }  // namespace
 
 void PublishKafka::initialize() {
@@ -235,20 +549,6 @@
   conn_.reset();
 }
 
-/**
- * Message delivery report callback using the richer rd_kafka_message_t object.
- */
-void PublishKafka::messageDeliveryCallback(rd_kafka_t* rk, const rd_kafka_message_t* rkmessage, void* /*opaque*/) {
-  if (rkmessage->_private == nullptr) {
-    return;
-  }
-  // allocated in PublishKafka::ReadCallback::produce
-  auto* func = reinterpret_cast<std::function<void(rd_kafka_t*, const rd_kafka_message_t*)>*>(rkmessage->_private);
-  try {
-    (*func)(rk, rkmessage);
-  } catch (...) { }
-  delete func;
-}
 
 bool PublishKafka::configureNewConnection(const std::shared_ptr<core::ProcessContext> &context) {
   std::string value;
@@ -256,14 +556,14 @@
   std::string valueConf;
   std::array<char, 512U> errstr{};
   rd_kafka_conf_res_t result;
-  const std::string PREFIX_ERROR_MSG = "PublishKafka: configure error result: ";
+  const char* const PREFIX_ERROR_MSG = "PublishKafka: configure error result: ";
 
   std::unique_ptr<rd_kafka_conf_t, rd_kafka_conf_deleter> conf_{ rd_kafka_conf_new() };
   if (conf_ == nullptr) {
     throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Failed to create rd_kafka_conf_t object");
   }
 
-  auto key = conn_->getKey();
+  const auto* const key = conn_->getKey();
 
   if (key->brokers_.empty()) {
     throw Exception(PROCESS_SCHEDULE_EXCEPTION, "There are no brokers");
@@ -451,7 +751,7 @@
   }
 
   // Set the delivery callback
-  rd_kafka_conf_set_dr_msg_cb(conf_.get(), &PublishKafka::messageDeliveryCallback);
+  rd_kafka_conf_set_dr_msg_cb(conf_.get(), &messageDeliveryCallback);
 
   // Set the logger callback
   rd_kafka_conf_set_log_cb(conf_.get(), &KafkaConnection::logCallback);
@@ -579,13 +879,13 @@
   }
   logger_->log_debug("Processing %lu flow files with a total size of %llu B", flowFiles.size(), actual_bytes);
 
-  auto messages = std::make_shared<Messages>();
+  auto messages = std::make_shared<Messages>(logger_);
   // We must add this to the messages set, so that it will be interrupted when notifyStop is called
   {
     std::lock_guard<std::mutex> lock(messages_mutex_);
     messages_set_.emplace(messages);
   }
-  // We also have to insure that it will be removed once we are done with it
+  // We also have to ensure that it will be removed once we are done with it
   const auto messagesSetGuard = gsl::finally([&]() {
     std::lock_guard<std::mutex> lock(messages_mutex_);
     messages_set_.erase(messages);
@@ -636,8 +936,8 @@
     bool failEmptyFlowFiles = true;
     context->getProperty(FailEmptyFlowFiles.getName(), failEmptyFlowFiles);
 
-    PublishKafka::ReadCallback callback(max_flow_seg_size_, kafkaKey, thisTopic->getTopic(), conn_->getConnection(), *flowFile,
-                                        attributeNameRegex_, messages, flow_file_index, failEmptyFlowFiles);
+    ReadCallback callback(max_flow_seg_size_, kafkaKey, thisTopic->getTopic(), conn_->getConnection(), *flowFile,
+                                        attributeNameRegex_, messages, flow_file_index, failEmptyFlowFiles, logger_);
     session->read(flowFile, &callback);
 
     if (!callback.called_) {
@@ -677,20 +977,20 @@
       for (size_t segment_num = 0; segment_num < flow_file.messages.size(); segment_num++) {
         const auto& message = flow_file.messages[segment_num];
         switch (message.status) {
-          case MessageStatus::MESSAGESTATUS_UNCOMPLETE:
+          case MessageStatus::InFlight:
             success = false;
             logger_->log_error("Waiting for delivery confirmation was interrupted for flow file %s segment %zu",
                 flowFiles[index]->getUUIDStr(),
                 segment_num);
           break;
-          case MessageStatus::MESSAGESTATUS_ERROR:
+          case MessageStatus::Error:
             success = false;
             logger_->log_error("Failed to deliver flow file %s segment %zu, error: %s",
                 flowFiles[index]->getUUIDStr(),
                 segment_num,
                 rd_kafka_err2str(message.err_code));
           break;
-          case MessageStatus::MESSAGESTATUS_SUCCESS:
+          case MessageStatus::Success:
             logger_->log_debug("Successfully delivered flow file %s segment %zu",
                 flowFiles[index]->getUUIDStr(),
                 segment_num);
diff --git a/extensions/librdkafka/PublishKafka.h b/extensions/librdkafka/PublishKafka.h
index 5d16717..7cc9101 100644
--- a/extensions/librdkafka/PublishKafka.h
+++ b/extensions/librdkafka/PublishKafka.h
@@ -50,36 +50,9 @@
 namespace minifi {
 namespace processors {
 
-#define COMPRESSION_CODEC_NONE "none"
-#define COMPRESSION_CODEC_GZIP "gzip"
-#define COMPRESSION_CODEC_SNAPPY "snappy"
-#define ROUND_ROBIN_PARTITIONING "Round Robin"
-#define RANDOM_PARTITIONING "Random Robin"
-#define USER_DEFINED_PARTITIONING "User-Defined"
-#define DELIVERY_REPLICATED "all"
-#define DELIVERY_ONE_NODE "1"
-#define DELIVERY_BEST_EFFORT "0"
-#define SECURITY_PROTOCOL_PLAINTEXT "plaintext"
-#define SECURITY_PROTOCOL_SSL "ssl"
-#define SECURITY_PROTOCOL_SASL_PLAINTEXT "sasl_plaintext"
-#define SECURITY_PROTOCOL_SASL_SSL "sasl_ssl"
-#define KAFKA_KEY_ATTRIBUTE "kafka.key"
-
 // PublishKafka Class
 class PublishKafka : public core::Processor {
  public:
-  // Constructor
-  /*!
-   * Create a new processor
-   */
-  explicit PublishKafka(std::string name, utils::Identifier uuid = utils::Identifier())
-      : core::Processor(std::move(name), uuid),
-        logger_(logging::LoggerFactory<PublishKafka>::getLogger()),
-        interrupted_(false) {
-  }
-
-  virtual ~PublishKafka() = default;
-
   static constexpr char const* ProcessorName = "PublishKafka";
 
   // Supported Properties
@@ -114,229 +87,14 @@
   static const core::Relationship Failure;
   static const core::Relationship Success;
 
-  // Message
-  enum class MessageStatus : uint8_t {
-    MESSAGESTATUS_UNCOMPLETE,
-    MESSAGESTATUS_ERROR,
-    MESSAGESTATUS_SUCCESS
-  };
-
-  struct MessageResult {
-    MessageStatus status = MessageStatus::MESSAGESTATUS_UNCOMPLETE;
-    rd_kafka_resp_err_t err_code = RD_KAFKA_RESP_ERR_UNKNOWN;
-  };
-
-  struct FlowFileResult {
-    bool flow_file_error = false;
-    std::vector<MessageResult> messages;
-  };
-
-  struct Messages {
-    std::mutex mutex;
-    std::condition_variable cv;
-    std::vector<FlowFileResult> flow_files;
-    bool interrupted = false;
-
-    void waitForCompletion() {
-      std::unique_lock<std::mutex> lock(mutex);
-      cv.wait(lock, [this]() -> bool {
-        if (interrupted) {
-          return true;
-        }
-        size_t index = 0U;
-        return std::all_of(this->flow_files.begin(), this->flow_files.end(), [&](const FlowFileResult& flow_file) {
-          index++;
-          if (flow_file.flow_file_error) {
-            return true;
-          }
-          return std::all_of(flow_file.messages.begin(), flow_file.messages.end(), [](const MessageResult& message) {
-            return message.status != MessageStatus::MESSAGESTATUS_UNCOMPLETE;
-          });
-        });
-      });
-    }
-
-    void modifyResult(size_t index, const std::function<void(FlowFileResult&)>& fun) {
-      std::unique_lock<std::mutex> lock(mutex);
-      fun(flow_files.at(index));
-      cv.notify_all();
-    }
-
-    size_t addFlowFile() {
-      std::lock_guard<std::mutex> lock(mutex);
-      flow_files.emplace_back();
-      return flow_files.size() - 1;
-    }
-
-    void iterateFlowFiles(const std::function<void(size_t /*index*/, const FlowFileResult& /*flow_file_result*/)>& fun) {
-      std::lock_guard<std::mutex> lock(mutex);
-      for (size_t index = 0U; index < flow_files.size(); index++) {
-        fun(index, flow_files[index]);
-      }
-    }
-
-    void interrupt() {
-      std::unique_lock<std::mutex> lock(mutex);
-      interrupted = true;
-      cv.notify_all();
-    }
-
-    bool wasInterrupted() {
-      std::lock_guard<std::mutex> lock(mutex);
-      return interrupted;
-    }
-  };
-
-  // Nest Callback Class for read stream
-  class ReadCallback : public InputStreamCallback {
-   public:
-    struct rd_kafka_headers_deleter {
-      void operator()(rd_kafka_headers_t* ptr) const noexcept {
-        rd_kafka_headers_destroy(ptr);
-      }
-    };
-
-    using rd_kafka_headers_unique_ptr = std::unique_ptr<rd_kafka_headers_t, rd_kafka_headers_deleter>;
-
-   private:
-    void allocate_message_object(const size_t segment_num) const {
-      messages_->modifyResult(flow_file_index_, [segment_num](FlowFileResult& flow_file) {
-        // allocate message object to be filled in by the callback in produce()
-        if (flow_file.messages.size() < segment_num + 1) {
-          flow_file.messages.resize(segment_num + 1);
-        }
-      });
-    }
-
-    static rd_kafka_headers_unique_ptr make_headers(const core::FlowFile& flow_file, utils::Regex& attribute_name_regex) {
-      const gsl::owner<rd_kafka_headers_t*> result{ rd_kafka_headers_new(8) };
-      if (!result) { throw std::bad_alloc{}; }
-
-      for (const auto& kv : flow_file.getAttributes()) {
-        if (attribute_name_regex.match(kv.first)) {
-          rd_kafka_header_add(result, kv.first.c_str(), kv.first.size(), kv.second.c_str(), kv.second.size());
-        }
-      }
-      return rd_kafka_headers_unique_ptr{ result };
-    }
-
-    rd_kafka_resp_err_t produce(const size_t segment_num, std::vector<unsigned char>& buffer, const size_t buflen) const {
-      const std::shared_ptr<Messages> messages_ptr_copy = this->messages_;
-      const auto flow_file_index_copy = this->flow_file_index_;
-      const auto produce_callback = [messages_ptr_copy, flow_file_index_copy, segment_num](rd_kafka_t * /*rk*/, const rd_kafka_message_t *rkmessage) {
-        messages_ptr_copy->modifyResult(flow_file_index_copy, [segment_num, rkmessage](FlowFileResult &flow_file) {
-          auto &message = flow_file.messages.at(segment_num);
-          message.err_code = rkmessage->err;
-          message.status = message.err_code == 0 ? MessageStatus::MESSAGESTATUS_SUCCESS : MessageStatus::MESSAGESTATUS_ERROR;
-        });
-      };
-      // release()d below, deallocated in PublishKafka::messageDeliveryCallback
-      auto callback_ptr = utils::make_unique<std::function<void(rd_kafka_t*, const rd_kafka_message_t*)>>(std::move(produce_callback));
-
-      allocate_message_object(segment_num);
-
-      const gsl::owner<rd_kafka_headers_t*> hdrs_copy = rd_kafka_headers_copy(hdrs.get());
-      const auto err = rd_kafka_producev(rk_, RD_KAFKA_V_RKT(rkt_), RD_KAFKA_V_PARTITION(RD_KAFKA_PARTITION_UA), RD_KAFKA_V_MSGFLAGS(RD_KAFKA_MSG_F_COPY), RD_KAFKA_V_VALUE(buffer.data(), buflen),
-                                         RD_KAFKA_V_HEADERS(hdrs_copy), RD_KAFKA_V_KEY(key_.c_str(), key_.size()), RD_KAFKA_V_OPAQUE(callback_ptr.get()), RD_KAFKA_V_END);
-      if (err == RD_KAFKA_RESP_ERR_NO_ERROR) {
-        // in case of failure, messageDeliveryCallback is not called and callback_ptr will delete the callback
-        // in case of success, messageDeliveryCallback takes ownership of the callback, so we no longer need to delete it
-        (void)callback_ptr.release();
-      } else {
-        // in case of failure, rd_kafka_producev doesn't take ownership of the headers, so we need to delete them
-        rd_kafka_headers_destroy(hdrs_copy);
-      }
-      return err;
-    }
-
-   public:
-    ReadCallback(const uint64_t max_seg_size,
-                 std::string key,
-                 rd_kafka_topic_t * const rkt,
-                 rd_kafka_t * const rk,
-                 const core::FlowFile& flowFile,
-                 utils::Regex &attributeNameRegex,
-                 std::shared_ptr<Messages> messages,
-                 const size_t flow_file_index,
-                 const bool fail_empty_flow_files)
-        : flow_size_(flowFile.getSize()),
-          max_seg_size_(max_seg_size == 0 || flow_size_ < max_seg_size ? flow_size_ : max_seg_size),
-          key_(std::move(key)),
-          rkt_(rkt),
-          rk_(rk),
-          hdrs(make_headers(flowFile, attributeNameRegex)),
-          messages_(std::move(messages)),
-          flow_file_index_(flow_file_index),
-          fail_empty_flow_files_(fail_empty_flow_files)
-    { }
-
-    int64_t process(const std::shared_ptr<io::BaseStream> stream) {
-      std::vector<unsigned char> buffer;
-
-      buffer.resize(max_seg_size_);
-      read_size_ = 0;
-      status_ = 0;
-      called_ = true;
-
-      assert(max_seg_size_ != 0 || flow_size_ == 0 && "max_seg_size_ == 0 implies flow_size_ == 0");
-      // ^^ therefore checking max_seg_size_ == 0 handles both division by zero and flow_size_ == 0 cases
-      const size_t reserved_msg_capacity = max_seg_size_ == 0 ? 1 : utils::intdiv_ceil(flow_size_, max_seg_size_);
-      messages_->modifyResult(flow_file_index_, [reserved_msg_capacity](FlowFileResult& flow_file) {
-        flow_file.messages.reserve(reserved_msg_capacity);
-      });
-
-      // If the flow file is empty, we still want to send the message, unless the user wants to fail_empty_flow_files_
-      if (flow_size_ == 0 && !fail_empty_flow_files_) {
-        produce(0, buffer, 0);
-        return 0;
-      }
-
-      for (size_t segment_num = 0; read_size_ < flow_size_; ++segment_num) {
-        const int readRet = stream->read(buffer.data(), buffer.size());
-        if (readRet < 0) {
-          status_ = -1;
-          error_ = "Failed to read from stream";
-          return read_size_;
-        }
-
-        if (readRet <= 0) { break; }
-
-        const auto err = produce(segment_num, buffer, readRet);
-        if (err) {
-          messages_->modifyResult(flow_file_index_, [segment_num, err](FlowFileResult& flow_file) {
-            auto& message = flow_file.messages.at(segment_num);
-            message.status = MessageStatus::MESSAGESTATUS_ERROR;
-            message.err_code = err;
-          });
-          status_ = -1;
-          error_ = rd_kafka_err2str(err);
-          return read_size_;
-        }
-        read_size_ += readRet;
-      }
-      return read_size_;
-    }
-
-    const uint64_t flow_size_ = 0;
-    const uint64_t max_seg_size_ = 0;
-    const std::string key_;
-    rd_kafka_topic_t * const rkt_ = nullptr;
-    rd_kafka_t * const rk_ = nullptr;
-    const rd_kafka_headers_unique_ptr hdrs;  // not null
-    const std::shared_ptr<Messages> messages_;
-    const size_t flow_file_index_;
-    int status_ = 0;
-    std::string error_;
-    int read_size_ = 0;
-    bool called_ = false;
-    const bool fail_empty_flow_files_ = true;
-  };
-
- public:
-  bool supportsDynamicProperties() override {
-    return true;
+  explicit PublishKafka(std::string name, utils::Identifier uuid = utils::Identifier())
+      : core::Processor(std::move(name), uuid) {
   }
 
+  ~PublishKafka() override = default;
+
+  bool supportsDynamicProperties() override { return true; }
+
   /**
    * Function that's executed when the processor is scheduled.
    * @param context process context.
@@ -348,25 +106,25 @@
   void onSchedule(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSessionFactory> &sessionFactory) override;
   void notifyStop() override;
 
+  class Messages;
+
  protected:
   bool configureNewConnection(const std::shared_ptr<core::ProcessContext> &context);
   bool createNewTopic(const std::shared_ptr<core::ProcessContext> &context, const std::string& topic_name);
 
  private:
-  static void messageDeliveryCallback(rd_kafka_t* rk, const rd_kafka_message_t* rkmessage, void* opaque);
-
-  std::shared_ptr<logging::Logger> logger_;
+  std::shared_ptr<logging::Logger> logger_{logging::LoggerFactory<PublishKafka>::getLogger()};
 
   KafkaConnectionKey key_;
   std::unique_ptr<KafkaConnection> conn_;
   std::mutex connection_mutex_;
 
-  uint32_t batch_size_;
-  uint64_t target_batch_payload_size_;
-  uint64_t max_flow_seg_size_;
+  uint32_t batch_size_{};
+  uint64_t target_batch_payload_size_{};
+  uint64_t max_flow_seg_size_{};
   utils::Regex attributeNameRegex_;
 
-  std::atomic<bool> interrupted_;
+  std::atomic<bool> interrupted_{false};
   std::mutex messages_mutex_;
   std::set<std::shared_ptr<Messages>> messages_set_;
 };
diff --git a/win_build_vs.bat b/win_build_vs.bat
index a2beefa..3a24778 100644
--- a/win_build_vs.bat
+++ b/win_build_vs.bat
@@ -64,6 +64,9 @@
     if [%%~x] EQU [/D] (
         set cmake_build_type=RelWithDebInfo
     )
+    if [%%~x] EQU [/DD] (
+        set cmake_build_type=Debug
+    )
     if [%%~x] EQU [/CI] (
         set "strict_gsl_checks=-DSTRICT_GSL_CHECKS=AUDIT"
     )