Repository navigation
[ISSUE #5413] Fix nine cluster ownership, takeover and subscription defects found by a new 15-case e2e suite - #5422
Merged
Merged
Conversation
…ation, address parsing and subscription publication Root cause (deregistration): namingDeregister looked the instance up with selectInstances(SERVICE, false), which returns an EMPTY list on the very client that registered the instance with nacos-client 2.1.1. The loop therefore matched nothing and leave() became a no-op: the stopped instance stayed in the live set until its heartbeat TTL expired (15s), so /session/recommend kept handing out its URL and PartitionAssigner kept spreading partitions over it. Fix: keep the Instance object this client registered and deregister that object directly; fall back to a listing scan (healthyOnly first, then all) for instances registered by a previous JVM, and expose createNamingService() as a protected seam for tests. Root cause (address parsing): the heartbeat value is "<ts>|<addr>[|<load...>]" but both parsers treated everything after the first '|' as the address, so with a LoadMeter wired the registration carried a bogus port and HttpForwarder built "http://host:port|0|0|0|0.42/internal/forward" (every cross-instance forward failed). Fix: strip at the second '|' in namingRegister and in ClusterMembership.instanceAddrOf (null when there is no separator at all). Root cause (subscription publication): one NamingService instance was published per subscription, but a nacos 2.1.1 client can publish only ONE instance per service — each new registration silently replaced the previous one, so with two subscribers the listing only ever contained the last one and peers dropped the other from their mirror (subscriber starves, no error). Verified against a live server: three registrations with distinct ports collapse into one instance, while the same three posted over the HTTP API all survive. Fix: publish ONE naming instance per runtime carrying its whole subscription set in the metadata (sub:<base64 key>=<base64 value>), re-register it on add/remove and deregister it when the last local subscription goes away; namingSubList() merges every runtime's metadata (legacy key/value shape still understood). refreshSubSnapshot() also no longer treats an empty listing (naming client reconnecting) as "every subscriber left". Tests: NacosMetaStoreDeregisterTest (3 cases, zero-dependency fake NamingService whose selectInstances(service,false) returns an empty list like 2.1.1).
poll()'s lazy-assign path called consumer.assign() with only the current topic's partitions. assign() replaces the consumer's WHOLE assignment, so every other topic the instance owns lost its assignment, and the records returned by that poll() whose topic did not match the requested one were dropped with "continue" even though the position had already advanced past them — permanent loss. Observed: "poll topic=OLD partition=0 rawRecords=1 bufferedPartitions=[NEW-0]". Fix: union-assign across every topic the instance already owns (same as assignPartitions), buffer each record under its OWN topic/partition, and track pullOffsets per record topic. Verified separately that assign(superset) keeps the positions of the already-assigned partitions, which is what makes the union assign safe.
… propagate stale-client eviction Graceful shutdown left every cluster subscription of the departing instance in the shared Meta (/em/subs/<topic>/<clientId>). Peers keep mirroring that prefix, so they kept forwarding events to the dead URL and kept polling/assigning partitions for a topic whose only subscriber no longer existed. Nothing could clean it up later: with nacos-client 2.1.1 only the REGISTERING client may deregister an ephemeral instance. Fix (1): UniIngressService.dropAllClusterSubscriptions() snapshots the locally subscribed client ids and runs each through unsubscribeByClient (local removal + cluster Meta delete + push-buffer cleanup); EventMeshApplication.shutdown() calls it after the servers and the session router are stopped and before the heartbeat scheduler stops / leave() runs, so the instance's own Meta client is still alive to perform the delete. Fix (2): the stale-client reaper (cleanupStaleClients, §13.6.5) called the inner SubscriptionManager directly, so eviction never reached the cluster Meta either — a SIGKILLed client stayed a ghost subscriber forever. It now goes through unsubscribeByClient() as well. Tests: UniIngressServiceTest covers both dropAllClusterSubscriptions paths (clustered and standalone); the e2e suite verifies that a stopped instance leaves no ghost entry, that a crashed client's entry disappears after the reaper runs, and that the cross-topic pull pollution this caused is gone.
…ne and stop taking them from a live peer
releaseStale() released a partition by CAS-ing the assignment record to an EMPTY string.
NacosMetaStore.tryAcquire is publishConfigCas, and the Nacos server rejects a CAS whose new
content is empty ("ErrCode:-400, ErrMsg:content invalid"), so the release never happened:
the record kept naming the old owner and the plan's new owner was fenced by a LIVE peer
forever. On scale-out the cluster deadlocked — A held [0], B [1], C [] out of a 2/1/1 plan
— and two partitions were then polled by nobody. The failure was swallowed at debug level.
Fix (1): RELEASED is now the non-empty sentinel "RELEASED". It still parses as "no owner"
(no '|' separator), so acquireOrFence treats it exactly like an absent record. A failed
release is retried on the next cycle and logged at WARN instead of being dropped with
myGen.
Fix (2): acquireOrFence no longer takes a partition away from a LIVE peer on token order.
FencingToken is seeded from the wall clock at boot, so a later-booted instance has
permanently higher tokens and would steal every partition — ownership froze on boot order
(4/0 instead of 2/2) and kept burning a CAS per cycle. A live peer now keeps the partition
until releaseStale hands it over; only a TTL-evicted (crashed) owner is taken over.
Tests: PartitionOwnershipConvergenceTest models a strict MetaStore that rejects empty
content (like the server) — it fails on the old empty tombstone with the exact deadlock
and passes now; PartitionFencingTest pins the CAS mechanics that keep the takeover safe.
enableCluster() built ClusterMembership with the instance id as its "address" and then scheduled the heartbeat with an initial delay of 0, while the advertised address is only resolved later in startupInternal(). The first beats therefore registered ip=<instanceId>, port=0 — an unroutable peer entry that /session/recommend and HttpForwarder resolve against (and that PartitionAssigner happily spreads partitions over). Fix: resolve the same address startupInternal() uses (-Deventmesh.http.advertisedAddr, else localIP:httpPort) and set it on ClusterMembership before the scheduler is created, so not a single beat can be published with the placeholder.
…te (15 cases) The existing e2e coverage only proved that a single client could publish and consume through a single runtime. Nothing exercised the cluster paths that actually break in production: partition ownership convergence, load-balanced delivery across two runtimes, takeover after a graceful stop or a SIGKILL, and the abnormal cases around duplicate subscribe, malformed requests and broker / meta-store outages. Those paths are exactly where the nine defects fixed in this branch were hiding. Extract the boot and probe scaffolding into ClusterE2EHarness so both suites share one definition of "instance is up", "partition owners converged" and "cluster subscription is mirrored", then add eleven cases on top of it: partitionOwnershipConvergesOnMembershipChanges loadBalanceModeSpreadsEventsAcrossSubscribersOnBothInstances samePartitionKeyAlwaysRoutesToTheSameSubscriber subscriberReconnectsToSurvivorAfterInstanceStopWithoutLoss clientsPinnedToStoppedInstanceAreTakenOverWithoutLoss duplicateSubscribeDoesNotDuplicateDelivery malformedRequestsAreRejectedAndServerStaysHealthy unsubscribeStopsDeliveryAndClearsTheClusterSubscription crashedClientSubscriptionIsEvictedFromClusterMeta brokerOutageLosesNothingThatWasAccepted (docker gated) metaStoreOutagePausesPollingThenRecoversWithoutLoss (docker gated) StaleClientChild is a standalone JVM client that subscribes and then sleeps forever, so the crash case can SIGKILL a real process instead of simulating a dropped connection in-process. Verified: 15/15 green against local em-e2e-nacos (8848) + em-e2e-kafka (9092) on the eventmesh-storage-kafka plugin with DeliveryTopology.PARTITION_OWNED_PULL.
…x too The fifteen e2e cases only ever ran on Kafka. The cluster layer is supposed to be broker-agnostic, but nothing proved it: partition ownership, takeover after a stop and cross-instance forwarding all touch backend-specific plumbing (queue counts, pull cursors, broker-assigned consumption), and a regression in any of the three plugins would have gone unnoticed. Introduce ClusterStorageBackend — the single place that knows how a backend is reached, which plugin implements it, how many partitions a topic has and whether EventMesh owns partitions at all — and make ClusterE2EHarness abstract over it. Each scenario suite becomes an abstract base (ClusterFailoverE2ESuite, ClusterBalanceE2ESuite) plus one thin concrete class per backend, so the bodies stay single-sourced instead of being copy-pasted three times: LoadBalancingAndFailoverE2EIntegrationTest ┐ ClusterFailoverRocketMQ4E2EIntegrationTest ├─ ClusterFailoverE2ESuite (4 cases) ClusterFailoverRocketMQ5E2EIntegrationTest ┘ ClusterBalanceAndResilienceE2EIntegrationTest ┐ ClusterBalanceRocketMQ4E2EIntegrationTest ├─ ClusterBalanceE2ESuite (11 cases) ClusterBalanceRocketMQ5E2EIntegrationTest ┘ The backend seam carries three non-obvious constraints: * The plugin is instantiated reflectively. e2eTest4 runs with storage-rocketmq5 excluded (5.5.0 moved common.protocol.* to remoting.protocol.*), so a static `new RocketMQ5RemotingStoragePlugin()` in the shared harness would drag that class into the 4.9 JVM. * Partition ownership is a backend property, not an EventMesh one. Kafka and RocketMQ 4.x assign client-side, so the /em/assignments/* table is the contract to assert. RocketMQ 5.x consumes with POP (broker-assigned, partitionCount == -1) and PartitionOwnership deliberately leaves the topic in poll-all mode — waiting on an assignment table there would hang forever. The ownership scenario therefore asserts the poll-all contract on that backend (no rows, delivery survives the same membership churn) and waitForTakeover() waits on the live set instead of on partitions, which is the honest evidence of takeover for broker-managed consumers. * The broker-outage case restarts the selected backend's container, so it stays meaningful on all three (em-e2e-kafka / em-rmq4-broker / em-rmq5-broker). Also harden the unsubscribe case: it published its baseline after a fixed 6s sleep, which races the CLUSTER_CYCLE_MS subscription mirror — an event pulled while only one of the two subscriptions is known routes to one subscriber and the baseline never completes (observed once as a 40s timeout on the RocketMQ 4.x run). It now waits until both cluster subscriptions are visible in the shared Meta before publishing. Verified against real infrastructure (local Nacos 2.1.1 + one broker at a time): Kafka (KRaft 3.7.0, 4 partitions) 15/15 pass · 235s RocketMQ 4 (4.9.5, 4 queues) 15/15 pass · 235s (failover 4/4, balance 11/11) RocketMQ 5 (5.5.0 POP, 4 queues) 15/15 pass · 332s Kafka and RocketMQ 5.x run in e2eTest5x; RocketMQ 4.x in the isolated-classpath e2eTest4. checkstyleMain/checkstyleTest clean.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Fixes #5413
Fixes #5414
Fixes #5415
Fixes #5416
Fixes #5417
Fixes #5418
Fixes #5419
Fixes #5420
Fixes #5421
Motivation
The cluster path of the v2 runtime (
DeliveryTopology.PARTITION_OWNED_PULLwith theNacos MetaStore +
eventmesh-storage-kafka) had no end-to-end coverage at all: theexisting e2e suites only proved that one client could publish and consume through one
runtime. Every path that actually matters in production — partition ownership
convergence on scale-out, load-balanced delivery across two runtimes, takeover after a
graceful stop and after a SIGKILL, and the abnormal cases around duplicate subscribe,
malformed requests, broker loss and meta-store loss — was untested, and nine real
defects were hiding there.
This PR adds a 15-case e2e suite that exercises those paths against a real Nacos 2.1.1
and a real broker — Kafka, RocketMQ 4.9.5 and RocketMQ 5.5.0 (the same 15 cases run
on each, 45 executions in total) — and fixes every defect the suite exposed. Each fix
has a regression test at the lowest level that can catch it (
NacosMetaStoreDeregisterTest,PartitionOwnershipConvergenceTest,UniIngressServiceTest) plus an e2e case thatreproduces the original failure.
The nine defects, grouped by symptom:
namingDeregisterlooks the instance up with
selectInstances(service, false), which — as ofnacos-client 2.1.1 — returns an empty list on the registering client itself, so the
deregister silently matched nothing and a stopped instance stayed in the live set
for the full 15s heartbeat TTL.
/session/recommendkept recommending it.topics share one consumer; the lazy
poll()path calledconsumer.assign()with asingle topic's partitions, which replaces the whole assignment, and records whose
topic did not match were dropped with the position already advanced — permanent loss.
ignores a deregister from a client other than the one that registered the ephemeral
instance, so
/em/subs/*for a stopping instance was never removed and every peerkept mirroring a ghost subscriber, polling dead topics and forwarding to dead URLs.
releaseStalereleased apartition by CAS-ing an empty value, and
publishConfigCasrejects empty content(
ErrCode:-400 content invalid). The record kept pointing at the old owner, so thenew owner was fenced forever and a third instance never got a partition.
value is
http://host:port|load|...; the old split producedhttp://host:port|0|0|0|0.42/...as a forward URL, so cross-instance forwarding to apeer was malformed.
acquireOrFencetook a partition away from a live peer whenever its own token was higher, and the
token is seeded from the boot clock, so the last instance to start stole every
partition (4/0 instead of 2/2).
cleanupStaleClientscalledsubscriptionManagerandpushServicedirectly and never propagated to the clusterMeta, so a crashed client's
/em/subs/*entry survived forever.([Bug] Cluster-wide subscription discovery exposes only one subscriber per runtime (naming instance collision) #5420). The store published one NamingService instance per subscription, but a
single nacos-client connection can only publish one instance per service in
2.1.1 — later registrations silently replaced earlier ones, so a runtime's own
subscriber could vanish as soon as a peer re-registered.
was created with the instance id as the address and a zero initial delay, so the
first registration was
ip=<instanceId>, port=0, racing the realsetSelfAddressinstartupInternal.Modifications
Product code
cluster/NacosMetaStore.javaInstanceobject on deregister instead of looking it up;fall back to a
healthyOnly=truescan when the map is empty.|so load fields are not part ofthe URL (metadata still carries the full value).
(
sub:<base64 topic>→<base64 clientId>, legacykey/valuestill read);namingSubList()merges every instance's metadata.refreshSubSnapshot()now refuses to overwrite a non-empty mirror with an emptylisting ("ignoring empty subscription listing — keeping N mirrored
subscription(s)"), which was the last source of subscription flapping.
createNamingService(Properties)seam for tests.cluster/PartitionOwnership.javaRELEASEDsentinel instead of an empty value; retry theCAS on the next cycle with a WARN and keep
myGenif it fails.ownerEvicted || currentToken == null); never preempt a live peer.cluster/ClusterMembership.java— same second-|truncation forinstanceAddrOf(returnsnullwhen the separator is missing).ingress/UniIngressService.java— newdropAllClusterSubscriptions(); stale-clienteviction now goes through
unsubscribeByClient()so local, cluster-Meta and pushcleanup all happen on one path.
boot/EventMeshApplication.java— resolve the advertised address(
eventmesh.http.advertisedAddr, defaulting to local IP + HTTP port) before theheartbeat scheduler is created; drop all cluster subscriptions on shutdown, after
traffic stops and while this instance's own Meta client is still alive.
storage-kafka/KafkaMeshStoragePlugin.java— union-assign every known topic in boththe lazy
poll()path andassignPartitions(), buffer byrecord.topic(), seek newpartitions, and key
pullOffsetsby topic.Tests
ClusterStorageBackend— the one place that knows how each backend is reached, whichplugin implements it, how many partitions a topic has, and whether EventMesh owns
partitions at all. Plugins are instantiated reflectively so the 4.x JVM never has to
resolve a 5.x class (see the note below on the classpath split).
ClusterE2EHarness— shared boot/probe scaffolding (instance boot, child JVMs,partitionOwners,clusterSubscription,recommendVia,waitFor,waitForTakeover,docker restart helpers), now parameterised by
backend().ClusterFailoverE2ESuite/ClusterBalanceE2ESuite— the scenario bodies, singlesourced, extended by one thin concrete class per backend (Kafka keeps the original
class names, so nothing that referenced them has to change).
ClusterBalanceE2ESuite— 11 cases: ownership convergence, load-balanced fan-out,partition-key affinity, takeover after a stop, takeover of pinned clients,
duplicate-subscribe idempotency, malformed-request robustness, unsubscribe clearing the
cluster Meta, crashed-client eviction, broker outage, MetaStore outage (the last two
docker-gated, and the broker one restarts whichever container belongs to the backend).
ClusterFailoverE2ESuite— 4 cases: recommend over the live set, overload avoidance,graceful restart switching, SIGKILL switching.
StaleClientChild— standalone JVM client that subscribes then sleeps, so the crashcase can SIGKILL a real process.
NacosMetaStoreDeregisterTest(3),PartitionOwnershipConvergenceTest(3, with astrict fake MetaStore that rejects empty content),
UniIngressServiceTest(+2),PartitionFencingTestcomment corrected to "pin CAS mechanics".build.gradle— the 4.x suites registered inBROKER_IT_4_CLASSES(isolatedclasspath), the 5.x suites in
BROKER_IT_5X_CLASSES.Why the plugin is loaded reflectively
e2eTest4runs witheventmesh-storage-rocketmq5excluded, because 4.9.5 and 5.5.0cannot share a classpath (5.x moved
common.protocol.*toremoting.protocol.*). Astatic
new RocketMQ5RemotingStoragePlugin()anywhere in the shared harness would dragthat class into the 4.x JVM, so the backend resolves its plugin with
Class.forNameinside the branch that actually needs it.
Partition ownership is a backend property, not an EventMesh one
Kafka and RocketMQ 4.x assign partitions client-side, so
/em/assignments/*is thecontract the ownership scenario asserts. RocketMQ 5.x consumes with POP: the broker
assigns messages,
partitionCountreturns-1, andPartitionOwnershipdeliberatelyleaves the topic in poll-all mode — waiting on an assignment table there would hang
forever. On that backend the same scenario asserts the poll-all contract instead (no
rows written, delivery survives identical membership churn), and
waitForTakeover()waits on the live set rather than on partitions, which is the honest evidence of
takeover for a broker-managed consumer.
Reviewing commit by commit is easiest: each commit maps to one or two issues, and the
last one is the test infrastructure plus the three-backend split.
Documentation
Nacos quirks and the release sentinel are documented in code comments and Javadoc)
bug fixes to existing cluster behaviour; observable behaviour is unchanged except
that it now matches what the docs already claim (ownership converges, no message
loss on takeover).
Verification
Real infrastructure on the dev box, one broker at a time, JDK 21: local Nacos 2.1.1
(8848), Kafka 3.7.0 KRaft (9092, 4 partitions), RocketMQ 4.9.5 (9876 / 10911, 4 queues),
RocketMQ 5.5.0 (19876 / 20911, 4 queues).
:eventmesh-runtime:testStateStoreDurabilityTest#twoInstancesConvergeOnSameView(unrelated to this change: it is an in-memory concurrency test whose 1s convergence budget loses under the CPU load of six running broker containers; it passes standalone, and CI — where the containers are not running — is green):eventmesh-storage-plugin:eventmesh-storage-kafka:testcheckstyleMain+checkstyleTestOn the RocketMQ 4.x run one case (
unsubscribeStopsDelivery...) timed out once becauseit published its baseline after a fixed 6s sleep, racing the 5s subscription mirror; the
case now waits on the observable cluster state instead of sleeping, and passed on every
subsequent run on all three backends.