You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "merlimat (via GitHub)" <gi...@apache.org> on 2026/10/07 20:45:04 UTC
[PR] [fix][broker] Never share a segment between members of a checkpoint consumer group [pulsar]
merlimat opened a new pull request, #26836:
URL: https://github.com/apache/pulsar/pull/26836
### Motivation
**Problem.** A `CheckpointConsumer` built with `.consumerGroup(...)` gets its segments from the topic's
controller. When the group has more members than the topic has segments, the controller shares a segment
among several members by entry-bucket (PIP-486), the same fan-out it does for a stream subscription, where
each sharer subscribes `Key_Shared` and the broker splits the buckets between them. A checkpoint consumer
has no shared subscription: it reads each assigned segment through its own `Reader`. Every member that
shares a segment therefore reads all of it, and each message is delivered once per sharer.
This breaks the `consumerGroup` contract ("each segment is assigned to exactly one consumer in the group at
a time") and the client specification: `spec/scalable-topics/client-api.md` §4.3 says "each unit is read by
exactly one member", and `key-shared.md` §9 says checkpoint consumers are unaffected by entry-bucketing.
A one-segment topic gets the default entry-bucket budget of 4, so a group of 2 to 4 members on a new topic
hits this unless auto scaling splits the segment first (it only splits a segment above the split-vs-rebucket
rate, 1,000 msg/s by default).
**Example.**
```java
// topic://public/default/orders has 1 segment (4 entry buckets)
for (int i = 0; i < 3; i++) {
client.newCheckpointConsumer(Schema.string())
.topic("topic://public/default/orders")
.consumerGroup("job")
.consumerName("member-" + i)
.startPosition(Checkpoint.earliest())
.create();
}
// Every message published to the topic is now received by all 3 members.
```
`SubscriptionCoordinator.computeAssignment` fans out whenever members outnumber segments, whatever the
consumer type:
```java
if (consumerCount <= segmentCount) {
// whole segments
} else {
// shared segment: every sharer gets the segment with its full bucket boundary list
}
```
and `ScalableCheckpointConsumer.applyAssignment` ignores the bucket ranges and opens a `Reader` on the whole
segment for each sharer:
```java
segmentReaders.computeIfAbsent(seg.segmentId(), id -> createSegmentReaderAsync(seg));
```
### Modifications
**Change.** A checkpoint consumer group always gets whole segments, one member each; members beyond the
number of segments stay idle. The coordinator learns that it serves a checkpoint group when a member
registers, before that registration computes the assignment. This also covers the controller-failover
restore path, where the consumer type is not persisted:
```java
// ScalableTopicController.registerConsumer
if (consumerType == ScalableConsumerType.STREAM) {
coordinator.installDrainChecker(this::isSegmentDrained);
} else if (consumerType == ScalableConsumerType.CHECKPOINT) {
coordinator.markCheckpointGroup();
}
// SubscriptionCoordinator.computeAssignment
if (consumerCount <= segmentCount || checkpointGroup) {
// whole segments, round-robin; surplus members get an empty assignment
```
Stream subscriptions are unchanged.
### Verifying this change
- [ ] Make sure that the change passes the CI checks.
This change added tests and can be verified as follows:
- `V5CheckpointConsumerGroupTest.testMembersOutnumberingSegmentsDeliverEachMessageOnce`: three members on a
one-segment topic; every message is delivered exactly once across the group, and exactly one member owns
the segment. It fails without the fix: each member receives all 100 messages.
- `SubscriptionCoordinatorTest.testCheckpointGroupNeverSharesSegment` and
`testRestoredCheckpointGroupNeverSharesSegmentOnReconnect` (controller failover).
- The existing `SubscriptionCoordinatorTest`, `V5CheckpointConsumerGroupTest`,
`V5CheckpointConsumerDagReplayTest` and `V5EntryBucketDispatchTest` (stream fan-out unchanged) pass.
### Does this pull request potentially affect one of the following parts:
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
*If the box was checked, please highlight the changes*
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Never share a segment between members of a checkpoint consumer group [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26836:
URL: https://github.com/apache/pulsar/pull/26836#discussion_r4212389905
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicController.java:
##########
@@ -893,9 +894,13 @@ public CompletableFuture<ConsumerAssignment> registerConsumer(String subscriptio
// The coordinator may have been created on the failover-restore path (consumer
// type unknown then; we defaulted to "no parent-drain enforcement"). Now that we
// know the type, upgrade if it's STREAM. installDrainChecker is a no-op if the
- // coordinator already has a checker, so safe to call unconditionally.
+ // coordinator already has a checker, so safe to call unconditionally. A CHECKPOINT
+ // group must be marked before its first assignment is computed: it never shares a
+ // segment between members.
if (consumerType == ScalableConsumerType.STREAM) {
coordinator.installDrainChecker(this::isSegmentDrained);
+ } else if (consumerType == ScalableConsumerType.CHECKPOINT) {
+ coordinator.markCheckpointGroup();
Review Comment:
**[QUALITY] failover wiring of markCheckpointGroup() is not pinned by a test**
The restore-path case this call exists for is only tested at the coordinator level, where `testRestoredCheckpointGroupNeverSharesSegmentOnReconnect` calls `markCheckpointGroup()` itself. The integration test covers a fresh coordinator, whose type is known at creation. So deleting this `else if` would keep the unit tests green and only break a controller failover. Optional: a controller-level test that restores a subscription (`restoreSubscription`), then registers a CHECKPOINT member and asserts the whole-segment assignment.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Never share a segment between members of a checkpoint consumer group [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat commented on code in PR #26836:
URL: https://github.com/apache/pulsar/pull/26836#discussion_r4212756846
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/scalable/ScalableTopicController.java:
##########
@@ -893,9 +894,13 @@ public CompletableFuture<ConsumerAssignment> registerConsumer(String subscriptio
// The coordinator may have been created on the failover-restore path (consumer
// type unknown then; we defaulted to "no parent-drain enforcement"). Now that we
// know the type, upgrade if it's STREAM. installDrainChecker is a no-op if the
- // coordinator already has a checker, so safe to call unconditionally.
+ // coordinator already has a checker, so safe to call unconditionally. A CHECKPOINT
+ // group must be marked before its first assignment is computed: it never shares a
+ // segment between members.
if (consumerType == ScalableConsumerType.STREAM) {
coordinator.installDrainChecker(this::isSegmentDrained);
+ } else if (consumerType == ScalableConsumerType.CHECKPOINT) {
+ coordinator.markCheckpointGroup();
Review Comment:
Added `ScalableTopicControllerTest.testRestoredCheckpointGroupGetsWholeSegments` in b360aac1034. It persists three members of a checkpoint group on a one-segment topic with 4 entry-buckets, restores them through `initialize()` (so the coordinator is created without a type), reconnects one as `CHECKPOINT` and checks it gets the segment whole. It fails without the `else if` in `registerConsumer`.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Never share a segment between members of a checkpoint consumer group [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat commented on PR #26836:
URL: https://github.com/apache/pulsar/pull/26836#issuecomment-6048318138
Thanks for the review and for checking the tests without the fix!
On the subscribe-response race: confirmed, and it can last longer than briefly. Since the response and the pushed update carry the same layout epoch, the client applies the late response last and keeps the stale assignment until the next rebalance; for a checkpoint group, two members read the same segment until then. Members starting together, such as a Flink job's subtasks, are the likely trigger. I'll fix it in a follow-up: the client needs a way to tell a membership rebalance from the initial assignment of the same layout epoch.
Agreed on the STREAM/CHECKPOINT name collision; not worth handling.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Never share a segment between members of a checkpoint consumer group [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat merged PR #26836:
URL: https://github.com/apache/pulsar/pull/26836
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org