lordgarrish commented on code in PR #286:
URL: https://github.com/apache/ignite-extensions/pull/286#discussion_r1796861596
##########
modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/KafkaToIgniteCdcStreamerApplier.java:
##########
@@ -260,7 +278,44 @@ private boolean
filterAndPossiblyUpdateMetadata(ConsumerRecord<Integer, byte[]>
metrics.incrementReceivedEvents();
- return F.isEmpty(caches) || caches.contains(rec.key());
+ return F.isEmpty(caches) || caches.contains(rec.key()) ||
matchesRegexTemplates(rec.key());
+ }
+
+ /**
+ * Gets caches names from CDC client and finds match
+ * between cache id and user's regex templates.
+ *
+ * @param key Cache id.
+ * @return True if match is found.
+ */
+ private boolean matchesRegexTemplates(Integer key) {
+ Optional<String> cache = streamer.getCaches().stream()
+ .filter(name -> CU.cacheId(name) == key)
+ .findAny();
+
+ Optional<String> matchedCache = cache.filter(this::matchesFilters);
+
+ matchedCache.ifPresent(c -> caches.add(CU.cacheId(c)));
Review Comment:
@NSAmelchev, can you explain please, on what condition `caches` could be
immutable? In
[AbstractKafkaToIgniteCdcStreamer](https://github.com/apache/ignite-extensions/blob/5b4f9de742927509c27eb36c13641f81fd0a986a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/AbstractKafkaToIgniteCdcStreamer.java#L161)
I see initialization of `caches` like this:
```
caches = streamerCfg.getCaches().stream()
.map(CU::cacheId).collect(Collectors.toSet());
```
Can `Collectors.toSet()` return the immutable set?
--
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: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
