You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by na...@apache.org on 2023/11/24 09:42:08 UTC
(ignite-extensions) branch master updated: IGNITE-20954 Prohibit the same group for event and metadata consumers (#246)
This is an automated email from the ASF dual-hosted git repository.
namelchev pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite-extensions.git
The following commit(s) were added to refs/heads/master by this push:
new 6419fd36 IGNITE-20954 Prohibit the same group for event and metadata consumers (#246)
6419fd36 is described below
commit 6419fd36d9bb9400699311b5665afc4c5d33d38a
Author: Nikita Amelchev <ns...@gmail.com>
AuthorDate: Fri Nov 24 12:42:03 2023 +0300
IGNITE-20954 Prohibit the same group for event and metadata consumers (#246)
---
.../kafka/AbstractKafkaToIgniteCdcStreamer.java | 8 +++-
.../ignite/cdc/kafka/KafkaToIgniteLoaderTest.java | 7 ++++
.../same-consumers-group/kafka-to-ignite.xml | 48 ++++++++++++++++++++++
.../loader/same-consumers-group/kafka.properties | 2 +
4 files changed, 63 insertions(+), 2 deletions(-)
diff --git a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/AbstractKafkaToIgniteCdcStreamer.java b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/AbstractKafkaToIgniteCdcStreamer.java
index 27de5274..2df4d204 100644
--- a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/AbstractKafkaToIgniteCdcStreamer.java
+++ b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/AbstractKafkaToIgniteCdcStreamer.java
@@ -107,8 +107,12 @@ abstract class AbstractKafkaToIgniteCdcStreamer implements Runnable {
appliers = new ArrayList<>(streamerCfg.getThreadCount());
runners = new ArrayList<>(streamerCfg.getThreadCount());
- if (!kafkaProps.containsKey(ConsumerConfig.GROUP_ID_CONFIG))
- throw new IllegalArgumentException("Kafka properties don't contains " + ConsumerConfig.GROUP_ID_CONFIG);
+ String grp = kafkaProps.getProperty(ConsumerConfig.GROUP_ID_CONFIG);
+
+ A.ensure(grp != null, "Kafka properties don't contains " + ConsumerConfig.GROUP_ID_CONFIG);
+ A.ensure(
+ !grp.equals(streamerCfg.getMetadataConsumerGroup()),
+ "The group of event and metadata consumers must be different.");
kafkaProps.put(KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class.getName());
kafkaProps.put(VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
diff --git a/modules/cdc-ext/src/test/java/org/apache/ignite/cdc/kafka/KafkaToIgniteLoaderTest.java b/modules/cdc-ext/src/test/java/org/apache/ignite/cdc/kafka/KafkaToIgniteLoaderTest.java
index cb2d385f..6938448b 100644
--- a/modules/cdc-ext/src/test/java/org/apache/ignite/cdc/kafka/KafkaToIgniteLoaderTest.java
+++ b/modules/cdc-ext/src/test/java/org/apache/ignite/cdc/kafka/KafkaToIgniteLoaderTest.java
@@ -44,6 +44,13 @@ public class KafkaToIgniteLoaderTest extends GridCommonAbstractTest {
"Spring bean with provided name doesn't exist"
);
+ assertThrows(
+ null,
+ () -> loadKafkaToIgniteStreamer("loader/same-consumers-group/kafka-to-ignite.xml"),
+ IllegalArgumentException.class,
+ "The group of event and metadata consumers must be different."
+ );
+
KafkaToIgniteCdcStreamer streamer = loadKafkaToIgniteStreamer("loader/kafka-to-ignite-correct.xml");
assertNotNull(streamer);
diff --git a/modules/cdc-ext/src/test/resources/loader/same-consumers-group/kafka-to-ignite.xml b/modules/cdc-ext/src/test/resources/loader/same-consumers-group/kafka-to-ignite.xml
new file mode 100644
index 00000000..55a6a9e0
--- /dev/null
+++ b/modules/cdc-ext/src/test/resources/loader/same-consumers-group/kafka-to-ignite.xml
@@ -0,0 +1,48 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+-->
+
+<beans xmlns="http://www.springframework.org/schema/beans"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xmlns:util="http://www.springframework.org/schema/util"
+ xsi:schemaLocation="
+ http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+ http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd">
+ <bean id="grid.cfg" class="org.apache.ignite.configuration.IgniteConfiguration">
+ <property name="dataStorageConfiguration">
+ <bean class="org.apache.ignite.configuration.DataStorageConfiguration">
+ <property name="defaultDataRegionConfiguration">
+ <bean class="org.apache.ignite.configuration.DataRegionConfiguration">
+ <property name="cdcEnabled" value="true" />
+ <property name="persistenceEnabled" value="true" />
+ </bean>
+ </property>
+ </bean>
+ </property>
+ </bean>
+
+ <bean id="streamer.cfg" class="org.apache.ignite.cdc.kafka.KafkaToIgniteCdcStreamerConfiguration">
+ <property name="topic" value="ignite" />
+ <property name="metadataTopic" value="ignite-metadata" />
+ <property name="kafkaPartsFrom" value="0" />
+ <property name="kafkaPartsTo" value="16" />
+ <property name="metadataConsumerGroup" value="group1" />
+ </bean>
+
+ <util:properties id="kafkaProperties" location="loader/same-consumers-group/kafka.properties" />
+</beans>
diff --git a/modules/cdc-ext/src/test/resources/loader/same-consumers-group/kafka.properties b/modules/cdc-ext/src/test/resources/loader/same-consumers-group/kafka.properties
new file mode 100644
index 00000000..346e21a7
--- /dev/null
+++ b/modules/cdc-ext/src/test/resources/loader/same-consumers-group/kafka.properties
@@ -0,0 +1,2 @@
+bootstrap.servers=127.0.0.1
+group.id=group1