Re: [PR] IGNITE-22530 CDC: Add regex filters for cache names [ignite-extensions]

["lordgarrish (via GitHub)" <gi...@apache.org>]

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-unsubscribe@ignite.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org