BewareMyPower commented on code in PR #601: URL: https://github.com/apache/pulsar-client-cpp/pull/601#discussion_r3576028893
########## lib/st/StProducerImpl.cc: ########## @@ -0,0 +1,460 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +#include "StProducerImpl.h" + +#include <pulsar/MessageBuilder.h> + +#include <algorithm> +#include <chrono> +#include <memory> +#include <string> +#include <unordered_set> +#include <utility> +#include <variant> +#include <vector> + +#include "MessageIdImpl.h" +#include "lib/ExecutorService.h" +#include "lib/LogUtils.h" + +DECLARE_LOG_OBJECT() + +namespace pulsar::st { + +StProducerImpl::StProducerImpl(pulsar::ClientImplPtr classic, ProducerConfig config) + : classic_(std::move(classic)), + config_(std::move(config)), + topic_(config_.topic), + producerName_(config_.producerName.value_or(std::string{})), + currentLayout_(std::make_shared<SegmentLayout>()) {} + +Future<void> StProducerImpl::start() { + dagWatch_ = std::make_shared<DagWatchSession>(classic_, config_.topic, /*createIfMissing*/ true); + std::weak_ptr<StProducerImpl> weak = weak_from_this(); + // Register the layout listener BEFORE start() so the very first layout is delivered to it. + dagWatch_->setLayoutChangeListener( + [weak](const SegmentLayout& newLayout, const SegmentLayout& oldLayout) { + if (auto self = weak.lock()) self->onLayoutChange(newLayout, oldLayout); + }); + dagWatch_->start().addListener([weak](const Expected<SegmentLayout>& result) { + if (auto self = weak.lock()) self->onStartResult(result); + }); + return startPromise_.getFuture(); +} + +void StProducerImpl::onStartResult(const Expected<SegmentLayout>& result) { + // Only the failure path is handled here: the DagWatchSession fails start()'s future (without + // invoking the layout listener) when the lookup fails before the first layout arrives. On + // success the listener runs immediately after and drives completeStart(). + if (!result) { + startPromise_.setError(result.error()); + } +} + +void StProducerImpl::onLayoutChange(const SegmentLayout& newLayout, const SegmentLayout& /*oldLayout*/) { + std::vector<Future<pulsar::Producer>> retired; + bool first = false; + { + std::lock_guard<std::mutex> lock(mutex_); + first = !sawFirstLayout_; + sawFirstLayout_ = true; + currentLayout_ = std::make_shared<SegmentLayout>(newLayout); + + std::unordered_set<std::uint64_t> active; + for (const auto& segment : newLayout.activeSegments()) active.insert(segment.segmentId); + for (auto it = segmentProducers_.begin(); it != segmentProducers_.end();) { + if (active.find(it->first) == active.end()) { + retired.push_back(std::move(it->second)); + it = segmentProducers_.erase(it); + } else { + ++it; + } + } + } + + // Close producers for segments that left the layout, fire-and-forget. + for (auto& future : retired) { + future.addListener([](const Expected<pulsar::Producer>& result) { + if (result) { + pulsar::Producer producer = *result; + producer.closeAsync([](pulsar::Result) {}); + } + }); + } + + if (first) { + completeStart(); + } else if (requiresExclusiveAttach(config_.accessMode)) { + // Exclusive modes claim newly-active segments eagerly (best-effort; failures are logged). + for (const auto& segment : newLayout.activeSegments()) { + getOrCreateSegmentProducerAsync(segment.segmentId); + } + } +} + +void StProducerImpl::completeStart() { + if (!requiresExclusiveAttach(config_.accessMode)) { + startPromise_.setSuccess(); // Shared: segment producers are created lazily on first send. + return; + } + auto layout = snapshotLayout(); + const auto& segments = layout->activeSegments(); + if (segments.empty()) { + startPromise_.setSuccess(); + return; + } + // Exclusive: eagerly attach every active segment; the create fails up front if the exclusive + // claim is refused, exactly like the Java strict eager attach. + auto remaining = std::make_shared<std::atomic<int>>(static_cast<int>(segments.size())); + for (const auto& segment : segments) { + getOrCreateSegmentProducerAsync(segment.segmentId) + .addListener([self = shared_from_this(), remaining](const Expected<pulsar::Producer>& result) { + if (!result) { + self->startPromise_.setError( + result.error()); // first error wins (complete is idempotent) + return; + } + if (remaining->fetch_sub(1) == 1) self->startPromise_.setSuccess(); + }); + } +} + +pulsar::ProducerConfiguration StProducerImpl::buildSegmentConfiguration(const Segment& segment) const { + // Build a FRESH config every time: pulsar::ProducerConfiguration's copy constructor shares its + // impl, so copy-then-mutate would clobber every other segment's config. + pulsar::ProducerConfiguration conf; + conf.setSchema(config_.schema); + conf.setAccessMode(toClassicAccessMode(config_.accessMode)); + if (config_.sendTimeoutMs) conf.setSendTimeout(static_cast<int>(*config_.sendTimeoutMs)); + conf.setBlockIfQueueFull(config_.blockIfQueueFull); + if (config_.initialSequenceId) conf.setInitialSequenceId(*config_.initialSequenceId); + for (const auto& [key, value] : config_.properties) conf.setProperty(key, value); + if (config_.producerName) { + conf.setProducerName(*config_.producerName + "-seg-" + std::to_string(segment.segmentId)); + } + // Legacy segments wrap an externally-managed persistent:// topic; mark the connection so the + // regular->scalable migration precheck (PIP-475) recognizes it. Real segment:// topics aren't. + if (segment.isLegacy()) { + conf.setProperty("__pulsar.v5.managed", "true"); + } + // PIP-486 entry-bucketing: only the encryption-disables-batching branch is portable today; the + // EntryBucketBatcherBuilder branch has no C++ seam yet (TODO when a batcher-builder lands). + if (conf.isEncryptionEnabled()) { + conf.setBatchingEnabled(false); + } + return conf; +} + +Future<pulsar::Producer> StProducerImpl::getOrCreateSegmentProducerAsync(std::uint64_t segmentId) { + std::lock_guard<std::mutex> lock(mutex_); + if (auto it = segmentProducers_.find(segmentId); it != segmentProducers_.end()) { + return it->second; // concurrent cold senders share the one creation attempt + } + + const Segment* segment = nullptr; + for (const auto& candidate : currentLayout_->activeSegments()) { + if (candidate.segmentId == segmentId) { + segment = &candidate; + break; + } + } + detail::Promise<pulsar::Producer> promise; + if (segment == nullptr) { + // Transient — don't cache, so a later routing attempt re-resolves against a fresh layout. + promise.setError(Error{ResultUnknownError, + "segment " + std::to_string(segmentId) + " is not in the active layout"}); + return promise.getFuture(); + } + + pulsar::ProducerConfiguration conf = buildSegmentConfiguration(*segment); + const std::string attachTopic = segment->attachTopicName(); + std::optional<std::string> assignedBrokerUrl; + if (const std::string* url = currentLayout_->brokerUrl(segmentId)) { + assignedBrokerUrl = *url; // pin to the DAG-provided owner broker Review Comment: The url here is always plaintext URL, when could `brokerUrlTls` be used? -- 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. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
