IGNITE-22670 Added KafkaToIgnite loader tests (#279)

apache · Jul 10, 2024 · e932e70 · e932e70
1 parent 5bd39e8
commit e932e70
Show file tree

Hide file tree

Showing 15 changed files with 749 additions and 0 deletions.
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
@@ -123,6 +123,62 @@ public void testLoadTimeoutProperties() throws Exception {
         });
     }
 
+    /** Tests setting kafka properties of kafka to ignite loaders. */
+    @Test
+    public void testKafkaProperties() {
+        Stream.of(
+            new String[] {
+                "loader/thin/kafka-to-ignite-client-without-topic.xml",
+                "Ouch! Argument cannot be null: Kafka topic"},
+            new String[] {
+                "loader/thin/kafka-to-ignite-client-without-metadata-topic.xml",
+                "Ouch! Argument cannot be null: Kafka metadata topic"},
+            new String[] {
+                "loader/kafka-to-ignite-without-topic.xml",
+                "Ouch! Argument cannot be null: Kafka topic"},
+            new String[] {
+                "loader/kafka-to-ignite-without-metadata-topic.xml",
+                "Ouch! Argument cannot be null: Kafka metadata topic"}
+        ).forEach(args -> assertThrows(null, () -> loadKafkaToIgniteStreamer(args[0]), NullPointerException.class, args[1]));
+
+        Stream.of(
+            new String[] {
+                "loader/thin/kafka-to-ignite-client-with-negative-partition-from.xml",
+                "Ouch! Argument is invalid: The Kafka partitions lower bound must be explicitly set to a value greater" +
+                    " than or equals to zero."},
+            new String[] {
+                "loader/thin/kafka-to-ignite-client-with-negative-partition-to.xml",
+                "Ouch! Argument is invalid: The Kafka partitions upper bound must be explicitly set to a value greater" +
+                    " than zero."},
+            new String[] {
+                "loader/thin/kafka-to-ignite-client-with-incorrect-partition-distribution.xml",
+                "Ouch! Argument is invalid: The Kafka partitions upper bound must be greater than lower bound."},
+            new String[] {
+                "loader/thin/kafka-to-ignite-client-with-negative-thread-count.xml",
+                "Ouch! Argument is invalid: Threads count value must me greater than zero."},
+            new String[] {
+                "loader/thin/kafka-to-ignite-client-with-incorrect-thread-count.xml",
+                "Ouch! Argument is invalid: Threads count must be less or equals to the total Kafka partitions count."},
+            new String[] {
+                "loader/kafka-to-ignite-with-negative-partition-from.xml",
+                "Ouch! Argument is invalid: The Kafka partitions lower bound must be explicitly set to a value greater" +
+                    " than or equals to zero."},
+            new String[] {
+                "loader/kafka-to-ignite-with-negative-partition-to.xml",
+                "Ouch! Argument is invalid: The Kafka partitions upper bound must be explicitly set to a value greater" +
+                    " than zero."},
+            new String[] {
+                "loader/kafka-to-ignite-with-incorrect-partition-distribution.xml",
+                "Ouch! Argument is invalid: The Kafka partitions upper bound must be greater than lower bound."},
+            new String[] {
+                "loader/kafka-to-ignite-with-negative-thread-count.xml",
+                "Ouch! Argument is invalid: Threads count value must me greater than zero."},
+            new String[] {
+                "loader/kafka-to-ignite-with-incorrect-thread-count.xml",
+                "Ouch! Argument is invalid: Threads count must be less or equals to the total Kafka partitions count."}
+        ).forEach(args -> assertThrows(null, () -> loadKafkaToIgniteStreamer(args[0]), IllegalArgumentException.class, args[1]));
+    }
+
     /** */
     @Test
     public void testInitSpringContextOnce() throws Exception {

diff --git a/...c-ext/src/test/resources/loader/kafka-to-ignite-with-incorrect-partition-distribution.xml b/...c-ext/src/test/resources/loader/kafka-to-ignite-with-incorrect-partition-distribution.xml
@@ -0,0 +1,52 @@
+<?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>
+
+    <util:constant static-field="org.apache.ignite.cdc.kafka.KafkaToIgniteLoaderTest.TEST_KAFKA_CONSUMER_POLL_TIMEOUT" id="consumerPollTimeout" />
+    <util:constant static-field="org.apache.ignite.cdc.kafka.KafkaToIgniteLoaderTest.TEST_KAFKA_REQUEST_TIMEOUT" id="requestTimeout" />
+
+    <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="5" />
+        <property name="kafkaPartsTo" value="4" />
+        <property name="kafkaRequestTimeout" ref="requestTimeout" />
+        <property name="kafkaConsumerPollTimeout" ref="consumerPollTimeout" />
+    </bean>
+
+    <util:properties id="kafkaProperties" location="loader/kafka.properties" />
+</beans>
diff --git a/modules/cdc-ext/src/test/resources/loader/kafka-to-ignite-with-incorrect-thread-count.xml b/modules/cdc-ext/src/test/resources/loader/kafka-to-ignite-with-incorrect-thread-count.xml
@@ -0,0 +1,53 @@
+<?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>
+
+    <util:constant static-field="org.apache.ignite.cdc.kafka.KafkaToIgniteLoaderTest.TEST_KAFKA_CONSUMER_POLL_TIMEOUT" id="consumerPollTimeout" />
+    <util:constant static-field="org.apache.ignite.cdc.kafka.KafkaToIgniteLoaderTest.TEST_KAFKA_REQUEST_TIMEOUT" id="requestTimeout" />
+
+    <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="kafkaRequestTimeout" ref="requestTimeout" />
+        <property name="kafkaConsumerPollTimeout" ref="consumerPollTimeout" />
+        <property name="threadCount" value="20" />
+    </bean>
+
+    <util:properties id="kafkaProperties" location="loader/kafka.properties" />
+</beans>
diff --git a/modules/cdc-ext/src/test/resources/loader/kafka-to-ignite-with-negative-partition-from.xml b/modules/cdc-ext/src/test/resources/loader/kafka-to-ignite-with-negative-partition-from.xml
@@ -0,0 +1,52 @@
+<?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>
+
+    <util:constant static-field="org.apache.ignite.cdc.kafka.KafkaToIgniteLoaderTest.TEST_KAFKA_CONSUMER_POLL_TIMEOUT" id="consumerPollTimeout" />
+    <util:constant static-field="org.apache.ignite.cdc.kafka.KafkaToIgniteLoaderTest.TEST_KAFKA_REQUEST_TIMEOUT" id="requestTimeout" />
+
+    <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="-1" />
+        <property name="kafkaPartsTo" value="16" />
+        <property name="kafkaRequestTimeout" ref="requestTimeout" />
+        <property name="kafkaConsumerPollTimeout" ref="consumerPollTimeout" />
+    </bean>
+
+    <util:properties id="kafkaProperties" location="loader/kafka.properties" />
+</beans>
diff --git a/modules/cdc-ext/src/test/resources/loader/kafka-to-ignite-with-negative-partition-to.xml b/modules/cdc-ext/src/test/resources/loader/kafka-to-ignite-with-negative-partition-to.xml
@@ -0,0 +1,52 @@
+<?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>
+
+    <util:constant static-field="org.apache.ignite.cdc.kafka.KafkaToIgniteLoaderTest.TEST_KAFKA_CONSUMER_POLL_TIMEOUT" id="consumerPollTimeout" />
+    <util:constant static-field="org.apache.ignite.cdc.kafka.KafkaToIgniteLoaderTest.TEST_KAFKA_REQUEST_TIMEOUT" id="requestTimeout" />
+
+    <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="-1" />
+        <property name="kafkaRequestTimeout" ref="requestTimeout" />
+        <property name="kafkaConsumerPollTimeout" ref="consumerPollTimeout" />
+    </bean>
+
+    <util:properties id="kafkaProperties" location="loader/kafka.properties" />
+</beans>
diff --git a/modules/cdc-ext/src/test/resources/loader/kafka-to-ignite-with-negative-thread-count.xml b/modules/cdc-ext/src/test/resources/loader/kafka-to-ignite-with-negative-thread-count.xml
@@ -0,0 +1,53 @@
+<?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>
+
+    <util:constant static-field="org.apache.ignite.cdc.kafka.KafkaToIgniteLoaderTest.TEST_KAFKA_CONSUMER_POLL_TIMEOUT" id="consumerPollTimeout" />
+    <util:constant static-field="org.apache.ignite.cdc.kafka.KafkaToIgniteLoaderTest.TEST_KAFKA_REQUEST_TIMEOUT" id="requestTimeout" />
+
+    <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="kafkaRequestTimeout" ref="requestTimeout" />
+        <property name="kafkaConsumerPollTimeout" ref="consumerPollTimeout" />
+        <property name="threadCount" value="-4" />
+    </bean>
+
+    <util:properties id="kafkaProperties" location="loader/kafka.properties" />
+</beans>