qqeasonchen opened a new pull request, #5422:
URL: https://github.com/apache/eventmesh/pull/5422
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_PULL`
with the
Nacos MetaStore + `eventmesh-storage-kafka`) had no end-to-end coverage at
all: the
existing 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 Kafka, 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 that
reproduces the original failure.
The nine defects, grouped by symptom:
1. **Graceful shutdown never deregisters the instance (#5413).**
`namingDeregister`
looks the instance up with `selectInstances(service, false)`, which — as
of
nacos-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/recommend` kept recommending it.
2. **Kafka storage loses messages when one instance owns several topics
(#5414).** All
topics share one consumer; the lazy `poll()` path called
`consumer.assign()` with a
single topic's partitions, which *replaces* the whole assignment, and
records whose
topic did not match were dropped with the position already advanced —
permanent loss.
3. **Graceful shutdown leaks cluster subscriptions (#5415).** Nacos 2.1.1
silently
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 peer
kept mirroring a ghost subscriber, polling dead topics and forwarding to
dead URLs.
4. **Partition ownership deadlocks after scale-out (#5416).** `releaseStale`
released a
partition by CAS-ing an *empty* value, and `publishConfigCas` rejects
empty content
(`ErrCode:-400 content invalid`). The record kept pointing at the old
owner, so the
new owner was fenced forever and a third instance never got a partition.
5. **Heartbeat load fields are parsed as part of the address (#5417).** The
membership
value is `http://host:port|load|...`; the old split produced
`http://host:port|0|0|0|0.42/...` as a forward URL, so cross-instance
forwarding to a
peer was malformed.
6. **Fencing-token order freezes ownership on boot order (#5418).**
`acquireOrFence`
took 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).
7. **Stale-client eviction is local-only (#5419).** `cleanupStaleClients`
called
`subscriptionManager` and `pushService` directly and never propagated to
the cluster
Meta, so a crashed client's `/em/subs/*` entry survived forever.
8. **Cluster-wide subscription discovery exposes only one subscriber per
runtime
(#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.
9. **First heartbeats publish a placeholder address (#5421).** The heartbeat
scheduler
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 real
`setSelfAddress` in `startupInternal`.
### Modifications
**Product code**
- `cluster/NacosMetaStore.java`
- Replay the registered `Instance` object on deregister instead of looking
it up;
fall back to a `healthyOnly=true` scan when the map is empty.
- Truncate the instance address at the *second* `|` so load fields are not
part of
the URL (metadata still carries the full value).
- Publish **one** naming instance per runtime carrying all subscriptions
in metadata
(`sub:<base64 topic>` → `<base64 clientId>`, legacy `key`/`value` still
read);
`namingSubList()` merges every instance's metadata.
- `refreshSubSnapshot()` now refuses to overwrite a non-empty mirror with
an empty
listing ("ignoring empty subscription listing — keeping N mirrored
subscription(s)"), which was the last source of subscription flapping.
- New `createNamingService(Properties)` seam for tests.
- `cluster/PartitionOwnership.java`
- Release with a non-empty `RELEASED` sentinel instead of an empty value;
retry the
CAS on the next cycle with a WARN and keep `myGen` if it fails.
- Only take a partition from an owner whose TTL expired (`ownerEvicted ||
currentToken == null`); never preempt a live peer.
- `cluster/ClusterMembership.java` — same second-`|` truncation for
`instanceAddrOf` (returns `null` when the separator is missing).
- `ingress/UniIngressService.java` — new `dropAllClusterSubscriptions()`;
stale-client
eviction now goes through `unsubscribeByClient()` so local, cluster-Meta
and push
cleanup all happen on one path.
- `boot/EventMeshApplication.java` — resolve the advertised address
(`eventmesh.http.advertisedAddr`, defaulting to local IP + HTTP port)
**before** the
heartbeat 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 both
the lazy `poll()` path and `assignPartitions()`, buffer by
`record.topic()`, seek new
partitions, and key `pullOffsets` by topic.
**Tests**
- `ClusterE2EHarness` — shared boot/probe scaffolding (instance boot, child
JVMs,
`partitionOwners`, `clusterSubscription`, `recommendVia`, `waitFor`,
docker restart
helpers); `LoadBalancingAndFailoverE2EIntegrationTest` was refactored onto
it.
- `ClusterBalanceAndResilienceE2EIntegrationTest` — 11 new 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, Kafka
outage, Nacos outage (the last two docker-gated).
- `StaleClientChild` — standalone JVM client that subscribes then sleeps, so
the crash
case can SIGKILL a real process.
- `NacosMetaStoreDeregisterTest` (3), `PartitionOwnershipConvergenceTest`
(3, with a
strict fake MetaStore that rejects empty content), `UniIngressServiceTest`
(+2),
`PartitionFencingTest` comment corrected to "pin CAS mechanics".
- `build.gradle` — both e2e classes registered in `BROKER_IT_5X_CLASSES`.
Reviewing commit by commit is easiest: each commit maps to one or two issues.
### Documentation
- Does this pull request introduce a new feature? (no)
- If yes, how is the feature documented? (not applicable — no new public
API; the
Nacos quirks and the release sentinel are documented in code comments and
Javadoc)
- If a feature is not applicable for documentation, explain why? — The
changes are
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
Run against local `em-e2e-nacos` (8848, standalone 2.1.1) and `em-e2e-kafka`
(9092,
KRaft, 4 partitions), JDK 21:
```
./gradlew :eventmesh-runtime:e2eTest5x --rerun-tasks \
-Dit.nacos=127.0.0.1:8848 -Dit.storage=kafka
-Dit.kafka.bootstrap=127.0.0.1:9092
```
| Check | Result |
|---|---|
| `ClusterBalanceAndResilienceE2EIntegrationTest` | 11/11 passed, 194.2s |
| `LoadBalancingAndFailoverE2EIntegrationTest` | 4/4 passed, 52.0s |
| full `e2eTest5x` | 15/15 passed, `BUILD SUCCESSFUL in 4m 10s` |
| `:eventmesh-runtime:test` | 349 tests, 0 failures |
| `:eventmesh-storage-plugin:eventmesh-storage-kafka:test` | 15 tests, 0
failures, 1 skipped |
| `checkstyleMain` + `checkstyleTest` | clean |
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
