Skip to content

Commit ef68eab

Browse files
jnmoynejnmoyne
committed
Improve elastic consumer group config validation.
Fix KV watchers being created with update only. Signed-off-by: Jean-Noël Moyne <jnmoyne@gmail.com>
1 parent 9ba10f5 commit ef68eab

4 files changed

Lines changed: 42 additions & 6 deletions

File tree

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -650,7 +650,7 @@ public void endOfData() {
650650
}
651651
};
652652

653-
watchSubscription = kv.watch(key, watcher, KeyValueWatchOption.UPDATES_ONLY);
653+
watchSubscription = kv.watch(key, watcher);
654654

655655
} catch (Exception e) {
656656
if (!stopped.get()) {

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

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -198,6 +198,10 @@ public void validate() throws ConsumerGroupException {
198198
}
199199
}
200200

201+
if (numWildcards == 0 && !">".equals(filterTokens[filterTokens.length - 1])) {
202+
throw new ConsumerGroupException("partitioning filters must have at least one * wildcard or end with > wildcard");
203+
}
204+
201205
if (partitioningWildcards != null && partitioningWildcards.length > numWildcards) {
202206
throw new ConsumerGroupException("the number of partitioning wildcards must not be larger than the total number of * wildcards in the filter");
203207
}
@@ -210,8 +214,8 @@ public void validate() throws ConsumerGroupException {
210214
}
211215
seenWildcards.add(pwc);
212216

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");
217+
if (pwc > numWildcards || pwc < 1) {
218+
throw new ConsumerGroupException("partitioning wildcard indexes must be between 1 and the number of * wildcards in the filter");
215219
}
216220
}
217221
}

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -440,7 +440,7 @@ public void endOfData() {
440440
}
441441
};
442442

443-
watchSubscription = kv.watch(key, watcher, KeyValueWatchOption.UPDATES_ONLY);
443+
watchSubscription = kv.watch(key, watcher);
444444

445445
} catch (Exception e) {
446446
if (!stopped.get()) {

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

Lines changed: 34 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -94,6 +94,27 @@ void testValidationMaxMembersZero() {
9494
assertTrue(exception.getMessage().contains("max number of members must be >= 1"));
9595
}
9696

97+
@Test
98+
void testValidationFilterNoWildcardNoGt() {
99+
ElasticConsumerGroupConfig config = new ElasticConsumerGroupConfig(
100+
4, "foo.bar", new int[]{}, 0, 0,
101+
Arrays.asList("m1", "m2"), new ArrayList<>()
102+
);
103+
104+
ConsumerGroupException exception = assertThrows(ConsumerGroupException.class, config::validate);
105+
assertTrue(exception.getMessage().contains("partitioning filters must have at least one * wildcard or end with > wildcard"));
106+
}
107+
108+
@Test
109+
void testValidationFilterEndingWithGtIsValid() {
110+
ElasticConsumerGroupConfig config = new ElasticConsumerGroupConfig(
111+
4, "foo.>", new int[]{}, 0, 0,
112+
Arrays.asList("m1", "m2"), new ArrayList<>()
113+
);
114+
115+
assertDoesNotThrow(config::validate);
116+
}
117+
97118
@Test
98119
void testValidationFilterNoWildcardWithWildcardsSpecified() {
99120
ElasticConsumerGroupConfig config = new ElasticConsumerGroupConfig(
@@ -102,7 +123,7 @@ void testValidationFilterNoWildcardWithWildcardsSpecified() {
102123
);
103124

104125
ConsumerGroupException exception = assertThrows(ConsumerGroupException.class, config::validate);
105-
assertTrue(exception.getMessage().contains("number of partitioning wildcards must not be larger than"));
126+
assertTrue(exception.getMessage().contains("partitioning filters must have at least one * wildcard or end with > wildcard"));
106127
}
107128

108129
@Test
@@ -154,7 +175,18 @@ void testValidationPartitioningWildcardsOutOfRange() {
154175
);
155176

156177
ConsumerGroupException exception = assertThrows(ConsumerGroupException.class, config::validate);
157-
assertTrue(exception.getMessage().contains("partitioning wildcard indexes must be less than or equal to"));
178+
assertTrue(exception.getMessage().contains("partitioning wildcard indexes must be between 1 and"));
179+
}
180+
181+
@Test
182+
void testValidationPartitioningWildcardsZeroIndex() {
183+
ElasticConsumerGroupConfig config = new ElasticConsumerGroupConfig(
184+
4, "foo.*", new int[]{0}, 0, 0,
185+
Arrays.asList("m1", "m2"), new ArrayList<>()
186+
);
187+
188+
ConsumerGroupException exception = assertThrows(ConsumerGroupException.class, config::validate);
189+
assertTrue(exception.getMessage().contains("partitioning wildcard indexes must be between 1 and"));
158190
}
159191

160192
@Test

0 commit comments

Comments
 (0)