tests/st/SegmentLayoutTest.cc - pulsar-client-cpp - Git at Google

 /**
  * 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 <gtest/gtest.h>

 #include <cstdint>
 #include <set>
 #include <string>

 #include "PulsarApi.pb.h"
 #include "lib/Murmur3_32Hash.h"
 #include "lib/st/SegmentLayout.h"

 using namespace pulsar::st;
 namespace proto = pulsar::proto;

 namespace {

 const std::string kTopic = "topic://public/default/orders";

 void addSegment(proto::ScalableTopicDAG& dag, std::uint64_t id, std::uint32_t start, std::uint32_t end,
                 proto::SegmentState state, const std::string& legacyTopic = {}) {
     auto* seg = dag.add_segments();
     seg->set_segment_id(id);
     seg->set_hash_start(start);
     seg->set_hash_end(end);
     seg->set_state(state);
     seg->set_created_at_epoch(1);
     seg->set_created_at_ms(1000);
     if (!legacyTopic.empty()) {
         seg->set_legacy_topic_name(legacyTopic);
     }
 }

 void addBroker(proto::ScalableTopicDAG& dag, std::uint64_t segmentId, const std::string& url,
                const std::string& urlTls = {}) {
     auto* addr = dag.add_segment_brokers();
     addr->set_segment_id(segmentId);
     addr->set_broker_url(url);
     if (!urlTls.empty()) {
         addr->set_broker_url_tls(urlTls);
     }
 }

 // A deterministic key whose 16-bit segment hash falls within [start, end].
 std::string findKeyInRange(std::uint32_t start, std::uint32_t end) {
     for (int i = 0; i < 1000000; i++) {
         std::string key = "key-" + std::to_string(i);
         std::uint32_t hash = ScalableTopicHashing::segmentHash(ScalableTopicHashing::murmur(key));
         if (hash >= start && hash <= end) {
             return key;
         }
     }
     ADD_FAILURE() << "no key found in range [" << start << ", " << end << "]";
     return {};
 }

 }  // namespace

 TEST(ScalableTopicHashingTest, testHashSplitsIntoIndependentHalves) {
     const std::string key = "some-routing-key";
     std::uint32_t murmur = ScalableTopicHashing::murmur(key);
     std::uint32_t segment = ScalableTopicHashing::segmentHash(murmur);
     std::uint32_t bucket = ScalableTopicHashing::entryBucketHash(murmur);
     ASSERT_LE(segment, 0xFFFFu);
     ASSERT_LE(bucket, 0xFFFFu);
     ASSERT_EQ((segment << 16) | bucket, murmur);
 }

 TEST(ScalableTopicHashingTest, testRawHashIsUnmasked) {
     // The classic makeHash clears bit 31; the raw hash must not, or the high
     // half of the split would be confined to [0, 0x7FFF]. Find a key where the
     // two differ, proving the raw path is in use.
     bool foundHighBit = false;
     for (int i = 0; i < 100000 && !foundHighBit; i++) {
         std::string key = "key-" + std::to_string(i);
         foundHighBit = (ScalableTopicHashing::murmur(key) & 0x80000000u) != 0;
     }
     ASSERT_TRUE(foundHighBit);
 }

 TEST(SegmentLayoutTest, testFromProtoBuildsSortedLayout) {
     proto::ScalableTopicDAG dag;
     dag.set_epoch(7);
     // Add out of order to verify sorting by range start.
     addSegment(dag, 2, 0x8000, 0xFFFF, proto::ACTIVE);
     addSegment(dag, 1, 0x0000, 0x7FFF, proto::ACTIVE);
     addSegment(dag, 0, 0x0000, 0xFFFF, proto::SEALED);
     addBroker(dag, 1, "pulsar://broker1:6650", "pulsar+ssl://broker1:6651");
     addBroker(dag, 2, "pulsar://broker2:6650");
     dag.set_controller_broker_url("pulsar://controller:6650");

     auto layout = SegmentLayout::fromProto(dag, kTopic);
     ASSERT_TRUE(layout);
     ASSERT_EQ(layout->epoch(), 7u);

     ASSERT_EQ(layout->activeSegments().size(), 2u);
     ASSERT_EQ(layout->activeSegments()[0].segmentId, 1u);  // sorted by range start
     ASSERT_EQ(layout->activeSegments()[1].segmentId, 2u);
     ASSERT_EQ(layout->activeSegments()[0].range, (HashRange{0x0000, 0x7FFF}));
     ASSERT_EQ(layout->activeSegments()[0].segmentTopicName, "segment://public/default/orders/0000-7fff-1");
     ASSERT_EQ(layout->activeSegments()[1].segmentTopicName, "segment://public/default/orders/8000-ffff-2");
     ASSERT_FALSE(layout->activeSegments()[0].isLegacy());
     ASSERT_EQ(layout->activeSegments()[0].attachTopicName(), layout->activeSegments()[0].segmentTopicName);

     ASSERT_EQ(layout->sealedSegments().size(), 1u);
     ASSERT_EQ(layout->sealedSegments()[0].segmentId, 0u);

     ASSERT_NE(layout->brokerUrl(1), nullptr);
     ASSERT_EQ(*layout->brokerUrl(1), "pulsar://broker1:6650");
     ASSERT_NE(layout->brokerUrlTls(1), nullptr);
     ASSERT_EQ(*layout->brokerUrlTls(1), "pulsar+ssl://broker1:6651");
     ASSERT_EQ(layout->brokerUrlTls(2), nullptr);
     ASSERT_EQ(layout->brokerUrl(42), nullptr);

     ASSERT_TRUE(layout->controllerBrokerUrl().has_value());
     ASSERT_EQ(*layout->controllerBrokerUrl(), "pulsar://controller:6650");
     ASSERT_FALSE(layout->controllerBrokerUrlTls().has_value());
 }

 TEST(SegmentLayoutTest, testFromProtoParsesLegacyAndBucketSplits) {
     proto::ScalableTopicDAG dag;
     dag.set_epoch(1);
     addSegment(dag, 0, 0x0000, 0xFFFF, proto::ACTIVE, "persistent://public/default/orders-partition-0");
     auto* seg = dag.mutable_segments(0);
     seg->add_entry_bucket_splits(0x4000);
     seg->add_entry_bucket_splits(0x8000);

     auto layout = SegmentLayout::fromProto(dag, kTopic);
     ASSERT_TRUE(layout);
     const Segment& segment = layout->activeSegments()[0];
     ASSERT_TRUE(segment.isLegacy());
     ASSERT_EQ(segment.attachTopicName(), "persistent://public/default/orders-partition-0");
     ASSERT_EQ(segment.entryBucketSplits, (std::vector<std::uint32_t>{0x4000, 0x8000}));
 }

 TEST(SegmentLayoutTest, testFromProtoRejectsBadTopicScheme) {
     proto::ScalableTopicDAG dag;
     dag.set_epoch(1);
     auto layout = SegmentLayout::fromProto(dag, "persistent://public/default/orders");
     ASSERT_FALSE(layout);
     ASSERT_EQ(layout.error().result, pulsar::ResultInvalidTopicName);
 }

 TEST(SegmentLayoutTest, testFromProtoRejectsMalformedHashRange) {
     {
         proto::ScalableTopicDAG dag;
         dag.set_epoch(1);
         addSegment(dag, 0, 0x8000, 0x7FFF, proto::ACTIVE);  // end < start
         ASSERT_FALSE(SegmentLayout::fromProto(dag, kTopic));
     }
     {
         proto::ScalableTopicDAG dag;
         dag.set_epoch(1);
         addSegment(dag, 0, 0x0000, 0x10000, proto::ACTIVE);  // end > 16-bit space
         ASSERT_FALSE(SegmentLayout::fromProto(dag, kTopic));
     }
 }

 namespace {

 Expected<SegmentLayout> fourSegmentLayout() {
     proto::ScalableTopicDAG dag;
     dag.set_epoch(1);
     addSegment(dag, 0, 0x0000, 0x3FFF, proto::ACTIVE);
     addSegment(dag, 1, 0x4000, 0x7FFF, proto::ACTIVE);
     addSegment(dag, 2, 0x8000, 0xBFFF, proto::ACTIVE);
     addSegment(dag, 3, 0xC000, 0xFFFF, proto::ACTIVE);
     return SegmentLayout::fromProto(dag, kTopic);
 }

 }  // namespace

 TEST(SegmentRouterTest, testKeyedRoutingHitsOwningRange) {
     auto layout = fourSegmentLayout();
     ASSERT_TRUE(layout);
     SegmentRouter router;

     for (int i = 0; i < 200; i++) {
         std::string key = "order-" + std::to_string(i);
         auto id = router.route(key, *layout);
         ASSERT_TRUE(id);
         std::uint32_t hash = ScalableTopicHashing::segmentHash(ScalableTopicHashing::murmur(key));
         const Segment& segment = layout->activeSegments()[*id];  // ids == sorted positions here
         ASSERT_TRUE(segment.range.contains(hash)) << "key " << key << " hash " << hash;
     }
 }

 TEST(SegmentRouterTest, testKeyedRoutingIsDeterministic) {
     auto layout = fourSegmentLayout();
     ASSERT_TRUE(layout);
     SegmentRouter router;
     for (int i = 0; i < 20; i++) {
         std::string key = "stable-key-" + std::to_string(i);
         auto first = router.route(key, *layout);
         auto second = router.route(key, *layout);
         ASSERT_TRUE(first);
         ASSERT_TRUE(second);
         ASSERT_EQ(*first, *second);
     }
 }

 TEST(SegmentRouterTest, testSingleFullRangeSegmentGetsEverything) {
     proto::ScalableTopicDAG dag;
     dag.set_epoch(1);
     addSegment(dag, 9, 0x0000, 0xFFFF, proto::ACTIVE);
     auto layout = SegmentLayout::fromProto(dag, kTopic);
     ASSERT_TRUE(layout);
     SegmentRouter router;
     for (int i = 0; i < 50; i++) {
         auto id = router.route("k" + std::to_string(i), *layout);
         ASSERT_TRUE(id);
         ASSERT_EQ(*id, 9u);
     }
 }

 TEST(SegmentRouterTest, testNoActiveSegmentsIsAnError) {
     SegmentLayout empty;
     SegmentRouter router;
     auto id = router.route("key", empty);
     ASSERT_FALSE(id);
     ASSERT_EQ(id.error().result, pulsar::ResultServiceUnitNotReady);
     auto rr = router.routeRoundRobin(empty);
     ASSERT_FALSE(rr);
 }

 TEST(SegmentRouterTest, testUncoveredHashIsAnError) {
     proto::ScalableTopicDAG dag;
     dag.set_epoch(1);
     addSegment(dag, 0, 0x0000, 0x7FFF, proto::ACTIVE);  // gap: [0x8000, 0xFFFF] uncovered
     auto layout = SegmentLayout::fromProto(dag, kTopic);
     ASSERT_TRUE(layout);
     SegmentRouter router;
     std::string uncoveredKey = findKeyInRange(0x8000, 0xFFFF);
     auto id = router.route(uncoveredKey, *layout);
     ASSERT_FALSE(id);
     ASSERT_EQ(id.error().result, pulsar::ResultUnknownError);
 }

 TEST(SegmentRouterTest, testAllLegacyRoutesModNLikeClassicPartitions) {
     proto::ScalableTopicDAG dag;
     dag.set_epoch(1);
     const int n = 4;
     for (int i = 0; i < n; i++) {
         addSegment(dag, i, i * 0x4000, (i + 1) * 0x4000 - 1, proto::ACTIVE,
                    "persistent://public/default/orders-partition-" + std::to_string(i));
     }
     auto layout = SegmentLayout::fromProto(dag, kTopic);
     ASSERT_TRUE(layout);
     SegmentRouter router;

     for (int i = 0; i < 100; i++) {
         std::string key = "legacy-key-" + std::to_string(i);
         auto id = router.route(key, *layout);
         ASSERT_TRUE(id);
         // Must match the classic partitioned-topic routing exactly.
         int expected = pulsar::Murmur3_32Hash().makeHash(key) % n;
         ASSERT_EQ(*id, static_cast<std::uint64_t>(expected)) << "key " << key;
     }
 }

 TEST(SegmentRouterTest, testMixedLegacyAndRegularUsesRangeRouting) {
     proto::ScalableTopicDAG dag;
     dag.set_epoch(2);
     addSegment(dag, 0, 0x0000, 0x7FFF, proto::ACTIVE, "persistent://public/default/orders-partition-0");
     addSegment(dag, 1, 0x8000, 0xFFFF, proto::ACTIVE);  // one regular segment -> range routing
     auto layout = SegmentLayout::fromProto(dag, kTopic);
     ASSERT_TRUE(layout);
     SegmentRouter router;

     std::string keyInUpperHalf = findKeyInRange(0x8000, 0xFFFF);
     auto id = router.route(keyInUpperHalf, *layout);
     ASSERT_TRUE(id);
     ASSERT_EQ(*id, 1u);
 }

 TEST(SegmentRouterTest, testRoundRobinCyclesAllActiveSegments) {
     auto layout = fourSegmentLayout();
     ASSERT_TRUE(layout);
     SegmentRouter router;
     std::set<std::uint64_t> seen;
     for (int i = 0; i < 8; i++) {
         auto id = router.routeRoundRobin(*layout);
         ASSERT_TRUE(id);
         seen.insert(*id);
     }
     ASSERT_EQ(seen.size(), 4u);  // every active segment hit
 }
	/**
	* 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 <gtest/gtest.h>

	#include <cstdint>
	#include <set>
	#include <string>

	#include "PulsarApi.pb.h"
	#include "lib/Murmur3_32Hash.h"
	#include "lib/st/SegmentLayout.h"

	using namespace pulsar::st;
	namespace proto = pulsar::proto;

	namespace {

	const std::string kTopic = "topic://public/default/orders";

	void addSegment(proto::ScalableTopicDAG& dag, std::uint64_t id, std::uint32_t start, std::uint32_t end,
	proto::SegmentState state, const std::string& legacyTopic = {}) {
	auto* seg = dag.add_segments();
	seg->set_segment_id(id);
	seg->set_hash_start(start);
	seg->set_hash_end(end);
	seg->set_state(state);
	seg->set_created_at_epoch(1);
	seg->set_created_at_ms(1000);
	if (!legacyTopic.empty()) {
	seg->set_legacy_topic_name(legacyTopic);
	}
	}

	void addBroker(proto::ScalableTopicDAG& dag, std::uint64_t segmentId, const std::string& url,
	const std::string& urlTls = {}) {
	auto* addr = dag.add_segment_brokers();
	addr->set_segment_id(segmentId);
	addr->set_broker_url(url);
	if (!urlTls.empty()) {
	addr->set_broker_url_tls(urlTls);
	}
	}

	// A deterministic key whose 16-bit segment hash falls within [start, end].
	std::string findKeyInRange(std::uint32_t start, std::uint32_t end) {
	for (int i = 0; i < 1000000; i++) {
	std::string key = "key-" + std::to_string(i);
	std::uint32_t hash = ScalableTopicHashing::segmentHash(ScalableTopicHashing::murmur(key));
	if (hash >= start && hash <= end) {
	return key;
	}
	}
	ADD_FAILURE() << "no key found in range [" << start << ", " << end << "]";
	return {};
	}

	} // namespace

	TEST(ScalableTopicHashingTest, testHashSplitsIntoIndependentHalves) {
	const std::string key = "some-routing-key";
	std::uint32_t murmur = ScalableTopicHashing::murmur(key);
	std::uint32_t segment = ScalableTopicHashing::segmentHash(murmur);
	std::uint32_t bucket = ScalableTopicHashing::entryBucketHash(murmur);
	ASSERT_LE(segment, 0xFFFFu);
	ASSERT_LE(bucket, 0xFFFFu);
	ASSERT_EQ((segment << 16) \| bucket, murmur);
	}

	TEST(ScalableTopicHashingTest, testRawHashIsUnmasked) {
	// The classic makeHash clears bit 31; the raw hash must not, or the high
	// half of the split would be confined to [0, 0x7FFF]. Find a key where the
	// two differ, proving the raw path is in use.
	bool foundHighBit = false;
	for (int i = 0; i < 100000 && !foundHighBit; i++) {
	std::string key = "key-" + std::to_string(i);
	foundHighBit = (ScalableTopicHashing::murmur(key) & 0x80000000u) != 0;
	}
	ASSERT_TRUE(foundHighBit);
	}

	TEST(SegmentLayoutTest, testFromProtoBuildsSortedLayout) {
	proto::ScalableTopicDAG dag;
	dag.set_epoch(7);
	// Add out of order to verify sorting by range start.
	addSegment(dag, 2, 0x8000, 0xFFFF, proto::ACTIVE);
	addSegment(dag, 1, 0x0000, 0x7FFF, proto::ACTIVE);
	addSegment(dag, 0, 0x0000, 0xFFFF, proto::SEALED);
	addBroker(dag, 1, "pulsar://broker1:6650", "pulsar+ssl://broker1:6651");
	addBroker(dag, 2, "pulsar://broker2:6650");
	dag.set_controller_broker_url("pulsar://controller:6650");

	auto layout = SegmentLayout::fromProto(dag, kTopic);
	ASSERT_TRUE(layout);
	ASSERT_EQ(layout->epoch(), 7u);

	ASSERT_EQ(layout->activeSegments().size(), 2u);
	ASSERT_EQ(layout->activeSegments()[0].segmentId, 1u); // sorted by range start
	ASSERT_EQ(layout->activeSegments()[1].segmentId, 2u);
	ASSERT_EQ(layout->activeSegments()[0].range, (HashRange{0x0000, 0x7FFF}));
	ASSERT_EQ(layout->activeSegments()[0].segmentTopicName, "segment://public/default/orders/0000-7fff-1");
	ASSERT_EQ(layout->activeSegments()[1].segmentTopicName, "segment://public/default/orders/8000-ffff-2");
	ASSERT_FALSE(layout->activeSegments()[0].isLegacy());
	ASSERT_EQ(layout->activeSegments()[0].attachTopicName(), layout->activeSegments()[0].segmentTopicName);

	ASSERT_EQ(layout->sealedSegments().size(), 1u);
	ASSERT_EQ(layout->sealedSegments()[0].segmentId, 0u);

	ASSERT_NE(layout->brokerUrl(1), nullptr);
	ASSERT_EQ(*layout->brokerUrl(1), "pulsar://broker1:6650");
	ASSERT_NE(layout->brokerUrlTls(1), nullptr);
	ASSERT_EQ(*layout->brokerUrlTls(1), "pulsar+ssl://broker1:6651");
	ASSERT_EQ(layout->brokerUrlTls(2), nullptr);
	ASSERT_EQ(layout->brokerUrl(42), nullptr);

	ASSERT_TRUE(layout->controllerBrokerUrl().has_value());
	ASSERT_EQ(*layout->controllerBrokerUrl(), "pulsar://controller:6650");
	ASSERT_FALSE(layout->controllerBrokerUrlTls().has_value());
	}

	TEST(SegmentLayoutTest, testFromProtoParsesLegacyAndBucketSplits) {
	proto::ScalableTopicDAG dag;
	dag.set_epoch(1);
	addSegment(dag, 0, 0x0000, 0xFFFF, proto::ACTIVE, "persistent://public/default/orders-partition-0");
	auto* seg = dag.mutable_segments(0);
	seg->add_entry_bucket_splits(0x4000);
	seg->add_entry_bucket_splits(0x8000);

	auto layout = SegmentLayout::fromProto(dag, kTopic);
	ASSERT_TRUE(layout);
	const Segment& segment = layout->activeSegments()[0];
	ASSERT_TRUE(segment.isLegacy());
	ASSERT_EQ(segment.attachTopicName(), "persistent://public/default/orders-partition-0");
	ASSERT_EQ(segment.entryBucketSplits, (std::vector<std::uint32_t>{0x4000, 0x8000}));
	}

	TEST(SegmentLayoutTest, testFromProtoRejectsBadTopicScheme) {
	proto::ScalableTopicDAG dag;
	dag.set_epoch(1);
	auto layout = SegmentLayout::fromProto(dag, "persistent://public/default/orders");
	ASSERT_FALSE(layout);
	ASSERT_EQ(layout.error().result, pulsar::ResultInvalidTopicName);
	}

	TEST(SegmentLayoutTest, testFromProtoRejectsMalformedHashRange) {
	{
	proto::ScalableTopicDAG dag;
	dag.set_epoch(1);
	addSegment(dag, 0, 0x8000, 0x7FFF, proto::ACTIVE); // end < start
	ASSERT_FALSE(SegmentLayout::fromProto(dag, kTopic));
	}
	{
	proto::ScalableTopicDAG dag;
	dag.set_epoch(1);
	addSegment(dag, 0, 0x0000, 0x10000, proto::ACTIVE); // end > 16-bit space
	ASSERT_FALSE(SegmentLayout::fromProto(dag, kTopic));
	}
	}

	namespace {

	Expected<SegmentLayout> fourSegmentLayout() {
	proto::ScalableTopicDAG dag;
	dag.set_epoch(1);
	addSegment(dag, 0, 0x0000, 0x3FFF, proto::ACTIVE);
	addSegment(dag, 1, 0x4000, 0x7FFF, proto::ACTIVE);
	addSegment(dag, 2, 0x8000, 0xBFFF, proto::ACTIVE);
	addSegment(dag, 3, 0xC000, 0xFFFF, proto::ACTIVE);
	return SegmentLayout::fromProto(dag, kTopic);
	}

	} // namespace

	TEST(SegmentRouterTest, testKeyedRoutingHitsOwningRange) {
	auto layout = fourSegmentLayout();
	ASSERT_TRUE(layout);
	SegmentRouter router;

	for (int i = 0; i < 200; i++) {
	std::string key = "order-" + std::to_string(i);
	auto id = router.route(key, *layout);
	ASSERT_TRUE(id);
	std::uint32_t hash = ScalableTopicHashing::segmentHash(ScalableTopicHashing::murmur(key));
	const Segment& segment = layout->activeSegments()[*id]; // ids == sorted positions here
	ASSERT_TRUE(segment.range.contains(hash)) << "key " << key << " hash " << hash;
	}
	}

	TEST(SegmentRouterTest, testKeyedRoutingIsDeterministic) {
	auto layout = fourSegmentLayout();
	ASSERT_TRUE(layout);
	SegmentRouter router;
	for (int i = 0; i < 20; i++) {
	std::string key = "stable-key-" + std::to_string(i);
	auto first = router.route(key, *layout);
	auto second = router.route(key, *layout);
	ASSERT_TRUE(first);
	ASSERT_TRUE(second);
	ASSERT_EQ(first, second);
	}
	}

	TEST(SegmentRouterTest, testSingleFullRangeSegmentGetsEverything) {
	proto::ScalableTopicDAG dag;
	dag.set_epoch(1);
	addSegment(dag, 9, 0x0000, 0xFFFF, proto::ACTIVE);
	auto layout = SegmentLayout::fromProto(dag, kTopic);
	ASSERT_TRUE(layout);
	SegmentRouter router;
	for (int i = 0; i < 50; i++) {
	auto id = router.route("k" + std::to_string(i), *layout);
	ASSERT_TRUE(id);
	ASSERT_EQ(*id, 9u);
	}
	}

	TEST(SegmentRouterTest, testNoActiveSegmentsIsAnError) {
	SegmentLayout empty;
	SegmentRouter router;
	auto id = router.route("key", empty);
	ASSERT_FALSE(id);
	ASSERT_EQ(id.error().result, pulsar::ResultServiceUnitNotReady);
	auto rr = router.routeRoundRobin(empty);
	ASSERT_FALSE(rr);
	}

	TEST(SegmentRouterTest, testUncoveredHashIsAnError) {
	proto::ScalableTopicDAG dag;
	dag.set_epoch(1);
	addSegment(dag, 0, 0x0000, 0x7FFF, proto::ACTIVE); // gap: [0x8000, 0xFFFF] uncovered
	auto layout = SegmentLayout::fromProto(dag, kTopic);
	ASSERT_TRUE(layout);
	SegmentRouter router;
	std::string uncoveredKey = findKeyInRange(0x8000, 0xFFFF);
	auto id = router.route(uncoveredKey, *layout);
	ASSERT_FALSE(id);
	ASSERT_EQ(id.error().result, pulsar::ResultUnknownError);
	}

	TEST(SegmentRouterTest, testAllLegacyRoutesModNLikeClassicPartitions) {
	proto::ScalableTopicDAG dag;
	dag.set_epoch(1);
	const int n = 4;
	for (int i = 0; i < n; i++) {
	addSegment(dag, i, i * 0x4000, (i + 1) * 0x4000 - 1, proto::ACTIVE,
	"persistent://public/default/orders-partition-" + std::to_string(i));
	}
	auto layout = SegmentLayout::fromProto(dag, kTopic);
	ASSERT_TRUE(layout);
	SegmentRouter router;

	for (int i = 0; i < 100; i++) {
	std::string key = "legacy-key-" + std::to_string(i);
	auto id = router.route(key, *layout);
	ASSERT_TRUE(id);
	// Must match the classic partitioned-topic routing exactly.
	int expected = pulsar::Murmur3_32Hash().makeHash(key) % n;
	ASSERT_EQ(*id, static_cast<std::uint64_t>(expected)) << "key " << key;
	}
	}

	TEST(SegmentRouterTest, testMixedLegacyAndRegularUsesRangeRouting) {
	proto::ScalableTopicDAG dag;
	dag.set_epoch(2);
	addSegment(dag, 0, 0x0000, 0x7FFF, proto::ACTIVE, "persistent://public/default/orders-partition-0");
	addSegment(dag, 1, 0x8000, 0xFFFF, proto::ACTIVE); // one regular segment -> range routing
	auto layout = SegmentLayout::fromProto(dag, kTopic);
	ASSERT_TRUE(layout);
	SegmentRouter router;

	std::string keyInUpperHalf = findKeyInRange(0x8000, 0xFFFF);
	auto id = router.route(keyInUpperHalf, *layout);
	ASSERT_TRUE(id);
	ASSERT_EQ(*id, 1u);
	}

	TEST(SegmentRouterTest, testRoundRobinCyclesAllActiveSegments) {
	auto layout = fourSegmentLayout();
	ASSERT_TRUE(layout);
	SegmentRouter router;
	std::set<std::uint64_t> seen;
	for (int i = 0; i < 8; i++) {
	auto id = router.routeRoundRobin(*layout);
	ASSERT_TRUE(id);
	seen.insert(*id);
	}
	ASSERT_EQ(seen.size(), 4u); // every active segment hit
	}