Skip to content

Commit 9ba10f5

Browse files
jnmoynejnmoyne
committed
Elastic consumer groups:
- enables not specifying any wildcard indexes, in which case the whole subject is used to compute the partition number. - Enables not specifying any partitioning filters, in which case ">" is used as the sole partition filter with the whole subject name being used to compute the partition number Port of synadia-io/orbit.go#46 Signed-off-by: Jean-Noël Moyne <jnmoyne@gmail.com>
1 parent 784885f commit 9ba10f5

5 files changed

Lines changed: 155 additions & 42 deletions

File tree

‎pcgroups/README.md‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ NATS Partitioned consumer groups come in two flavors: *elastic* and *static*.
2727

2828
***Static*** partitioned consumer groups assume that the stream already has a partition number present as the first token of the message's subjects (something that can be done automatically when messages are stored into to the stream by setting a subject transform for the stream). You can only create and delete static consumer groups. Any change to the consumer group's config in the KV bucket will cause all the member instances for all members of the group to stop consuming.
2929

30-
***Elastic*** partitioned consumer groups on the other hand are implemented differently: the stream doesn't need to already contain a partition number subject token and you can administratively add and drop members from the consumer group's config whenever you want without having to delete and re-create the consumer (like you have to with static consumer groups).
30+
***Elastic*** partitioned consumer groups on the other hand are implemented differently: the stream doesn't need to already contain a partition number subject token and you can administratively add and drop members from the consumer group's config whenever you want without having to delete and re-create the consumer (like you have to with static consumer groups). You have the option of specifying a subject filter for the consumer group and calculating the partition number from the subject name using a consistent hashing algorithm. Either through the use of `*` wildcards in the partitioning filter(s) and then specifying in the partitioning wildcards array the indexes of the `*` wildcards in the filter that you want to use for computing the partition number (you can specify between one index and all of the indexes), or by leaving that array of wildcard indexes empty (or not specifying a partitioning filter at all) in which case the partition number is calculated using the entirety of the message's subject.
3131

3232
***In both cases***
3333
In both cases you must specify when creating the consumer group the maximum number of members for the group (which is actually the number of partitions used when partitioning the messages), plus a list of "members" (named instances of the consuming application). The library takes care of distributing the members over the list of partitions using either a 'balanced' distribution (the partitions are evenly distributed between the members) or 'mappings' (where you assign administratively the mappings of partitions to the members). The membership list or mappings must be specified once at consumer group creation time for static consumer groups, but can be changed at any time for elastic consumer groups.
@@ -65,8 +65,8 @@ This `cg` CLI tool can be used by passing it commands and arguments directly, or
6565
For more details on the CLI visit the [Partitioned Consumer Groups CLI Project](https://github.com/synadia-io/orbit.java/tree/main/pcgroups-cli)
6666

6767
### Binaries
68-
You can download the latest `cg.jar` archived in a tar file
69-
[cg.tar](https://github.com/synadia-io/orbit.java/releases/download/pcgcli%2F0.1.0/cg.tar)
68+
You can download the latest `cg.jar` archived in a tar file
69+
[cg.tar](https://github.com/synadia-io/orbit.java/releases/download/pcgcli%2F0.1.0/cg.tar)
7070
or zip file
7171
[cg.zip](https://github.com/synadia-io/orbit.java/releases/download/pcgcli%2F0.1.0/cg.zip)
7272

@@ -117,4 +117,4 @@ Partitioned consumer groups require NATS server version 2.11 or above.
117117

118118
---
119119
Copyright (c) 2025 Synadia Communications Inc. All Rights Reserved.
120-
See [LICENSE](LICENSE) and [NOTICE](NOTICE) file for details.
120+
See [LICENSE](LICENSE) and [NOTICE](NOTICE) file for details.

‎pcgroups/src/main/java/io/synadia/pcg/ElasticConsumerGroup.java‎

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -111,6 +111,7 @@ public static ElasticConsumerGroupConfig create(Connection nc, String streamName
111111

112112
// Create the work queue stream with subject transform
113113
String workQueueStreamName = composeCGSName(streamName, consumerGroupName);
114+
String effectiveFilter = (filter != null && !filter.isEmpty()) ? filter : ">";
114115
String filterDest = getPartitioningTransformDest(config);
115116

116117
StreamConfiguration.Builder scBuilder = StreamConfiguration.builder()
@@ -129,12 +130,11 @@ public static ElasticConsumerGroupConfig create(Connection nc, String streamName
129130
}
130131

131132
// Add source with subject transform
132-
External external = null; // Local stream
133133
scBuilder.addSource(Source.builder()
134134
.sourceName(streamName)
135135
.startSeq(0)
136136
.subjectTransforms(SubjectTransform.builder()
137-
.source(filter)
137+
.source(effectiveFilter)
138138
.destination(filterDest)
139139
.build())
140140
.build());
@@ -534,14 +534,16 @@ private static ElasticConsumerGroupConfig getConfigFromKV(KeyValue kv, String st
534534
}
535535

536536
private static String getPartitioningTransformDest(ElasticConsumerGroupConfig config) {
537+
String effectiveFilter = (config.getFilter() != null && !config.getFilter().isEmpty()) ? config.getFilter() : ">";
537538
int[] wildcards = config.getPartitioningWildcards();
539+
538540
StringBuilder wildcardList = new StringBuilder();
539541
for (int i = 0; i < wildcards.length; i++) {
540542
if (i > 0) wildcardList.append(",");
541543
wildcardList.append(wildcards[i]);
542544
}
543545

544-
String[] filterTokens = config.getFilter().split("\\.");
546+
String[] filterTokens = effectiveFilter.split("\\.");
545547
int cwIndex = 1;
546548
for (int i = 0; i < filterTokens.length; i++) {
547549
if (filterTokens[i].equals("*")) {
@@ -551,6 +553,11 @@ private static String getPartitioningTransformDest(ElasticConsumerGroupConfig co
551553
}
552554

553555
String destFromFilter = String.join(".", filterTokens);
556+
557+
if (wildcards.length == 0) {
558+
return "{{Partition(" + config.getMaxMembers() + ")}}." + destFromFilter;
559+
}
560+
554561
return "{{Partition(" + config.getMaxMembers() + "," + wildcardList + ")}}." + destFromFilter;
555562
}
556563

‎pcgroups/src/main/java/io/synadia/pcg/ElasticConsumerGroupConfig.java‎

Lines changed: 32 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -189,35 +189,31 @@ public void validate() throws ConsumerGroupException {
189189
}
190190

191191
// Validate filter and partitioning wildcards
192-
if (filter == null || filter.isEmpty()) {
193-
throw new ConsumerGroupException("filter must not be empty");
194-
}
195-
196-
String[] filterTokens = filter.split("\\.");
197-
int numWildcards = 0;
198-
for (String token : filterTokens) {
199-
if ("*".equals(token)) {
200-
numWildcards++;
192+
if (filter != null && !filter.isEmpty()) {
193+
String[] filterTokens = filter.split("\\.");
194+
int numWildcards = 0;
195+
for (String token : filterTokens) {
196+
if ("*".equals(token)) {
197+
numWildcards++;
198+
}
201199
}
202-
}
203-
204-
if (numWildcards < 1) {
205-
throw new ConsumerGroupException("filter must contain at least one * wildcard");
206-
}
207-
208-
if (partitioningWildcards == null || partitioningWildcards.length < 1 || partitioningWildcards.length > numWildcards) {
209-
throw new ConsumerGroupException("the number of partitioning wildcards must be between 1 and the total number of * wildcards in the filter");
210-
}
211200

212-
Set<Integer> seenWildcards = new HashSet<>();
213-
for (int pwc : partitioningWildcards) {
214-
if (seenWildcards.contains(pwc)) {
215-
throw new ConsumerGroupException("partitioning wildcard indexes must be unique");
201+
if (partitioningWildcards != null && partitioningWildcards.length > numWildcards) {
202+
throw new ConsumerGroupException("the number of partitioning wildcards must not be larger than the total number of * wildcards in the filter");
216203
}
217-
seenWildcards.add(pwc);
218204

219-
if (pwc < 1 || pwc > numWildcards) {
220-
throw new ConsumerGroupException("partitioning wildcard indexes must be greater than 1 and less than or equal to the number of * wildcards in the filter");
205+
Set<Integer> seenWildcards = new HashSet<>();
206+
if (partitioningWildcards != null) {
207+
for (int pwc : partitioningWildcards) {
208+
if (seenWildcards.contains(pwc)) {
209+
throw new ConsumerGroupException("partitioning wildcard indexes must be unique");
210+
}
211+
seenWildcards.add(pwc);
212+
213+
if (pwc > numWildcards) {
214+
throw new ConsumerGroupException("partitioning wildcard indexes must be less than or equal to the number of * wildcards in the filter");
215+
}
216+
}
221217
}
222218
}
223219

@@ -266,15 +262,18 @@ public void validate() throws ConsumerGroupException {
266262
* Generates the subject transform destination for partitioning.
267263
*/
268264
public String getPartitioningTransformDest() {
265+
String effectiveFilter = (filter != null && !filter.isEmpty()) ? filter : ">";
266+
int[] effectiveWildcards = (partitioningWildcards != null) ? partitioningWildcards : new int[0];
267+
269268
StringBuilder wildcardList = new StringBuilder();
270-
for (int i = 0; i < partitioningWildcards.length; i++) {
269+
for (int i = 0; i < effectiveWildcards.length; i++) {
271270
if (i > 0) {
272271
wildcardList.append(",");
273272
}
274-
wildcardList.append(partitioningWildcards[i]);
273+
wildcardList.append(effectiveWildcards[i]);
275274
}
276275

277-
String[] filterTokens = filter.split("\\.");
276+
String[] filterTokens = effectiveFilter.split("\\.");
278277
int cwIndex = 1;
279278
for (int i = 0; i < filterTokens.length; i++) {
280279
if ("*".equals(filterTokens[i])) {
@@ -284,6 +283,11 @@ public String getPartitioningTransformDest() {
284283
}
285284

286285
String destFromFilter = String.join(".", filterTokens);
286+
287+
if (effectiveWildcards.length == 0) {
288+
return "{{Partition(" + maxMembers + ")}}." + destFromFilter;
289+
}
290+
287291
return "{{Partition(" + maxMembers + "," + wildcardList + ")}}." + destFromFilter;
288292
}
289293

‎pcgroups/src/test/java/io/synadia/pcg/ElasticConsumerGroupTest.java‎

Lines changed: 59 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -95,25 +95,44 @@ void testValidationMaxMembersZero() {
9595
}
9696

9797
@Test
98-
void testValidationFilterNoWildcard() {
98+
void testValidationFilterNoWildcardWithWildcardsSpecified() {
9999
ElasticConsumerGroupConfig config = new ElasticConsumerGroupConfig(
100100
4, "foo.bar", new int[]{1}, 0, 0,
101101
Arrays.asList("m1", "m2"), new ArrayList<>()
102102
);
103103

104104
ConsumerGroupException exception = assertThrows(ConsumerGroupException.class, config::validate);
105-
assertTrue(exception.getMessage().contains("filter must contain at least one * wildcard"));
105+
assertTrue(exception.getMessage().contains("number of partitioning wildcards must not be larger than"));
106106
}
107107

108108
@Test
109-
void testValidationPartitioningWildcardsEmpty() {
109+
void testValidationPartitioningWildcardsEmptyIsValid() {
110110
ElasticConsumerGroupConfig config = new ElasticConsumerGroupConfig(
111111
4, "foo.*", new int[]{}, 0, 0,
112112
Arrays.asList("m1", "m2"), new ArrayList<>()
113113
);
114114

115-
ConsumerGroupException exception = assertThrows(ConsumerGroupException.class, config::validate);
116-
assertTrue(exception.getMessage().contains("number of partitioning wildcards must be between"));
115+
assertDoesNotThrow(config::validate);
116+
}
117+
118+
@Test
119+
void testValidationNoFilterIsValid() {
120+
ElasticConsumerGroupConfig config = new ElasticConsumerGroupConfig(
121+
4, null, new int[]{}, 0, 0,
122+
Arrays.asList("m1", "m2"), new ArrayList<>()
123+
);
124+
125+
assertDoesNotThrow(config::validate);
126+
}
127+
128+
@Test
129+
void testValidationEmptyFilterIsValid() {
130+
ElasticConsumerGroupConfig config = new ElasticConsumerGroupConfig(
131+
4, "", new int[]{}, 0, 0,
132+
Arrays.asList("m1", "m2"), new ArrayList<>()
133+
);
134+
135+
assertDoesNotThrow(config::validate);
117136
}
118137

119138
@Test
@@ -124,7 +143,7 @@ void testValidationPartitioningWildcardsTooMany() {
124143
);
125144

126145
ConsumerGroupException exception = assertThrows(ConsumerGroupException.class, config::validate);
127-
assertTrue(exception.getMessage().contains("number of partitioning wildcards must be between"));
146+
assertTrue(exception.getMessage().contains("number of partitioning wildcards must not be larger than"));
128147
}
129148

130149
@Test
@@ -135,7 +154,7 @@ void testValidationPartitioningWildcardsOutOfRange() {
135154
);
136155

137156
ConsumerGroupException exception = assertThrows(ConsumerGroupException.class, config::validate);
138-
assertTrue(exception.getMessage().contains("partitioning wildcard indexes must be greater than 1"));
157+
assertTrue(exception.getMessage().contains("partitioning wildcard indexes must be less than or equal to"));
139158
}
140159

141160
@Test
@@ -232,6 +251,39 @@ void testGetPartitioningTransformDestPartialWildcards() {
232251
assertEquals("{{Partition(8,2)}}.a.{{Wildcard(1)}}.b.{{Wildcard(2)}}.c.{{Wildcard(3)}}", dest);
233252
}
234253

254+
@Test
255+
void testGetPartitioningTransformDestNoWildcards() {
256+
ElasticConsumerGroupConfig config = new ElasticConsumerGroupConfig(
257+
4, "foo.*", new int[]{}, 0, 0,
258+
new ArrayList<>(), new ArrayList<>()
259+
);
260+
261+
String dest = config.getPartitioningTransformDest();
262+
assertEquals("{{Partition(4)}}.foo.{{Wildcard(1)}}", dest);
263+
}
264+
265+
@Test
266+
void testGetPartitioningTransformDestNoFilter() {
267+
ElasticConsumerGroupConfig config = new ElasticConsumerGroupConfig(
268+
4, null, new int[]{}, 0, 0,
269+
new ArrayList<>(), new ArrayList<>()
270+
);
271+
272+
String dest = config.getPartitioningTransformDest();
273+
assertEquals("{{Partition(4)}}.>", dest);
274+
}
275+
276+
@Test
277+
void testGetPartitioningTransformDestEmptyFilter() {
278+
ElasticConsumerGroupConfig config = new ElasticConsumerGroupConfig(
279+
4, "", new int[]{}, 0, 0,
280+
new ArrayList<>(), new ArrayList<>()
281+
);
282+
283+
String dest = config.getPartitioningTransformDest();
284+
assertEquals("{{Partition(4)}}.>", dest);
285+
}
286+
235287
@Test
236288
void testJsonSerializationWithMembers() throws JsonParseException {
237289
ElasticConsumerGroupConfig config = new ElasticConsumerGroupConfig(

‎pcgroups/src/test/java/io/synadia/pcg/IntegrationTest.java‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -243,6 +243,56 @@ void testElastic() throws Exception {
243243

244244
// Delete elastic consumer group
245245
ElasticConsumerGroup.delete(nc, streamName, cgName);
246+
247+
// --- Test elastic with no partitioning filters (partition on whole subject) ---
248+
String cgName2 = "group2";
249+
AtomicInteger c3 = new AtomicInteger(0);
250+
AtomicInteger c4 = new AtomicInteger(0);
251+
252+
// Create elastic consumer group with no filter (null) and empty wildcards
253+
ElasticConsumerGroup.create(nc, streamName, cgName2, 2, null,
254+
new int[]{}, -1, -1);
255+
256+
// Start consuming on both members
257+
ConsumerGroupConsumeContext cc3 = ElasticConsumerGroup.consume(nc, streamName, cgName2, "m1", msg -> {
258+
c3.incrementAndGet();
259+
try {
260+
msg.ack();
261+
} catch (Exception e) {
262+
throw new RuntimeException(e);
263+
}
264+
}, config);
265+
266+
ConsumerGroupConsumeContext cc4 = ElasticConsumerGroup.consume(nc, streamName, cgName2, "m2", msg -> {
267+
c4.incrementAndGet();
268+
try {
269+
msg.ack();
270+
} catch (Exception e) {
271+
throw new RuntimeException(e);
272+
}
273+
}, config);
274+
275+
// Add members
276+
ElasticConsumerGroup.addMembers(nc, streamName, cgName2, Arrays.asList("m1", "m2"));
277+
278+
// Wait for all 30 messages (from previous publishes) to be consumed, split between the 2 members
279+
// The stream has 30 messages total (10 + 10 + 10 from the three publish phases above)
280+
deadline = System.currentTimeMillis() + 10000;
281+
while (c3.get() + c4.get() < 30) {
282+
Thread.sleep(100);
283+
if (System.currentTimeMillis() > deadline) {
284+
fail("timeout no-filter elastic: c3=" + c3.get() + " c4=" + c4.get() + " expected total=30");
285+
}
286+
}
287+
288+
assertEquals(30, c3.get() + c4.get());
289+
290+
cc3.stop();
291+
cc4.stop();
292+
293+
// Delete elastic consumer group
294+
ElasticConsumerGroup.delete(nc, streamName, cgName2);
295+
246296
nc.close();
247297
}
248298
}

0 commit comments

Comments
 (0)