You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sa...@apache.org on 2020/05/31 20:26:55 UTC
[ignite] branch master updated: IGNITE-12359 Migrate RocketMQ
module to ignite-extensions - Fixes #7809.
This is an automated email from the ASF dual-hosted git repository.
samaitra pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 0139490 IGNITE-12359 Migrate RocketMQ module to ignite-extensions - Fixes #7809.
0139490 is described below
commit 01394900b36d5ae0ec1d23486cf3b7940b6da9be
Author: samaitra <sa...@gmail.com>
AuthorDate: Sun May 31 15:26:19 2020 -0500
IGNITE-12359 Migrate RocketMQ module to ignite-extensions - Fixes #7809.
Signed-off-by: samaitra <sa...@gmail.com>
---
assembly/libs/README.txt | 1 -
modules/rocketmq/README.txt | 25 ---
modules/rocketmq/licenses/apache-2.0.txt | 202 -------------------
modules/rocketmq/pom.xml | 94 ---------
.../ignite/stream/rocketmq/RocketMQStreamer.java | 151 ---------------
.../ignite/stream/rocketmq/package-info.java | 21 --
.../stream/rocketmq/RocketMQStreamerTest.java | 215 ---------------------
.../stream/rocketmq/RocketMQStreamerTestSuite.java | 31 ---
.../ignite/stream/rocketmq/TestRocketMQServer.java | 148 --------------
.../ignite/stream/rocketmq/package-info.java | 21 --
parent/pom.xml | 5 -
pom.xml | 1 -
12 files changed, 915 deletions(-)
diff --git a/assembly/libs/README.txt b/assembly/libs/README.txt
index 1365cc1..fc40956 100644
--- a/assembly/libs/README.txt
+++ b/assembly/libs/README.txt
@@ -95,7 +95,6 @@ The following modules are available:
- ignite-osgi-karaf (to seemlessly intall ignite into Apache Karaf container)
- ignite-osgi-paxlogging (to expose PAX Logging API to Log4j if needed)
- ignite-rest-http (for HTTP REST messages)
-- ignite-rocketmq (for streaming messages from Apache RocketMQ into Ignite)
- ignite-scalar (for ignite Scala API)
- ignite-scalar_2.10 (for Ignite Scala 2.10 API)
- ignite-schedule (for Cron-based task scheduling)
diff --git a/modules/rocketmq/README.txt b/modules/rocketmq/README.txt
deleted file mode 100644
index 55a117b..0000000
--- a/modules/rocketmq/README.txt
+++ /dev/null
@@ -1,25 +0,0 @@
-Apache Ignite RocketMQ Streamer Module
---------------------------------------
-
-Apache Ignite RocketMQ Streamer module provides streaming from RocketMQ to Ignite cache.
-
-To use Ignite RocketMQ Streamer module, first import it to your Maven project.
-
-<project xmlns="http://maven.apache.org/POM/4.0.0"
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
- http://maven.apache.org/xsd/maven-4.0.0.xsd">
- ...
- <dependencies>
- ...
- <dependency>
- <groupId>org.apache.ignite</groupId>
- <artifactId>ignite-rocketmq</artifactId>
- <version>${ignite.version}</version>
- </dependency>
- ...
- </dependencies>
- ...
-</project>
-
-Then, initialize and start it as, for instance, done in RocketMQStreamerTest.java.
diff --git a/modules/rocketmq/licenses/apache-2.0.txt b/modules/rocketmq/licenses/apache-2.0.txt
deleted file mode 100644
index d645695..0000000
--- a/modules/rocketmq/licenses/apache-2.0.txt
+++ /dev/null
@@ -1,202 +0,0 @@
-
- Apache License
- Version 2.0, January 2004
- http://www.apache.org/licenses/
-
- TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
-
- 1. Definitions.
-
- "License" shall mean the terms and conditions for use, reproduction,
- and distribution as defined by Sections 1 through 9 of this document.
-
- "Licensor" shall mean the copyright owner or entity authorized by
- the copyright owner that is granting the License.
-
- "Legal Entity" shall mean the union of the acting entity and all
- other entities that control, are controlled by, or are under common
- control with that entity. For the purposes of this definition,
- "control" means (i) the power, direct or indirect, to cause the
- direction or management of such entity, whether by contract or
- otherwise, or (ii) ownership of fifty percent (50%) or more of the
- outstanding shares, or (iii) beneficial ownership of such entity.
-
- "You" (or "Your") shall mean an individual or Legal Entity
- exercising permissions granted by this License.
-
- "Source" form shall mean the preferred form for making modifications,
- including but not limited to software source code, documentation
- source, and configuration files.
-
- "Object" form shall mean any form resulting from mechanical
- transformation or translation of a Source form, including but
- not limited to compiled object code, generated documentation,
- and conversions to other media types.
-
- "Work" shall mean the work of authorship, whether in Source or
- Object form, made available under the License, as indicated by a
- copyright notice that is included in or attached to the work
- (an example is provided in the Appendix below).
-
- "Derivative Works" shall mean any work, whether in Source or Object
- form, that is based on (or derived from) the Work and for which the
- editorial revisions, annotations, elaborations, or other modifications
- represent, as a whole, an original work of authorship. For the purposes
- of this License, Derivative Works shall not include works that remain
- separable from, or merely link (or bind by name) to the interfaces of,
- the Work and Derivative Works thereof.
-
- "Contribution" shall mean any work of authorship, including
- the original version of the Work and any modifications or additions
- to that Work or Derivative Works thereof, that is intentionally
- submitted to Licensor for inclusion in the Work by the copyright owner
- or by an individual or Legal Entity authorized to submit on behalf of
- the copyright owner. For the purposes of this definition, "submitted"
- means any form of electronic, verbal, or written communication sent
- to the Licensor or its representatives, including but not limited to
- communication on electronic mailing lists, source code control systems,
- and issue tracking systems that are managed by, or on behalf of, the
- Licensor for the purpose of discussing and improving the Work, but
- excluding communication that is conspicuously marked or otherwise
- designated in writing by the copyright owner as "Not a Contribution."
-
- "Contributor" shall mean Licensor and any individual or Legal Entity
- on behalf of whom a Contribution has been received by Licensor and
- subsequently incorporated within the Work.
-
- 2. Grant of Copyright License. Subject to the terms and conditions of
- this License, each Contributor hereby grants to You a perpetual,
- worldwide, non-exclusive, no-charge, royalty-free, irrevocable
- copyright license to reproduce, prepare Derivative Works of,
- publicly display, publicly perform, sublicense, and distribute the
- Work and such Derivative Works in Source or Object form.
-
- 3. Grant of Patent License. Subject to the terms and conditions of
- this License, each Contributor hereby grants to You a perpetual,
- worldwide, non-exclusive, no-charge, royalty-free, irrevocable
- (except as stated in this section) patent license to make, have made,
- use, offer to sell, sell, import, and otherwise transfer the Work,
- where such license applies only to those patent claims licensable
- by such Contributor that are necessarily infringed by their
- Contribution(s) alone or by combination of their Contribution(s)
- with the Work to which such Contribution(s) was submitted. If You
- institute patent litigation against any entity (including a
- cross-claim or counterclaim in a lawsuit) alleging that the Work
- or a Contribution incorporated within the Work constitutes direct
- or contributory patent infringement, then any patent licenses
- granted to You under this License for that Work shall terminate
- as of the date such litigation is filed.
-
- 4. Redistribution. You may reproduce and distribute copies of the
- Work or Derivative Works thereof in any medium, with or without
- modifications, and in Source or Object form, provided that You
- meet the following conditions:
-
- (a) You must give any other recipients of the Work or
- Derivative Works a copy of this License; and
-
- (b) You must cause any modified files to carry prominent notices
- stating that You changed the files; and
-
- (c) You must retain, in the Source form of any Derivative Works
- that You distribute, all copyright, patent, trademark, and
- attribution notices from the Source form of the Work,
- excluding those notices that do not pertain to any part of
- the Derivative Works; and
-
- (d) If the Work includes a "NOTICE" text file as part of its
- distribution, then any Derivative Works that You distribute must
- include a readable copy of the attribution notices contained
- within such NOTICE file, excluding those notices that do not
- pertain to any part of the Derivative Works, in at least one
- of the following places: within a NOTICE text file distributed
- as part of the Derivative Works; within the Source form or
- documentation, if provided along with the Derivative Works; or,
- within a display generated by the Derivative Works, if and
- wherever such third-party notices normally appear. The contents
- of the NOTICE file are for informational purposes only and
- do not modify the License. You may add Your own attribution
- notices within Derivative Works that You distribute, alongside
- or as an addendum to the NOTICE text from the Work, provided
- that such additional attribution notices cannot be construed
- as modifying the License.
-
- You may add Your own copyright statement to Your modifications and
- may provide additional or different license terms and conditions
- for use, reproduction, or distribution of Your modifications, or
- for any such Derivative Works as a whole, provided Your use,
- reproduction, and distribution of the Work otherwise complies with
- the conditions stated in this License.
-
- 5. Submission of Contributions. Unless You explicitly state otherwise,
- any Contribution intentionally submitted for inclusion in the Work
- by You to the Licensor shall be under the terms and conditions of
- this License, without any additional terms or conditions.
- Notwithstanding the above, nothing herein shall supersede or modify
- the terms of any separate license agreement you may have executed
- with Licensor regarding such Contributions.
-
- 6. Trademarks. This License does not grant permission to use the trade
- names, trademarks, service marks, or product names of the Licensor,
- except as required for reasonable and customary use in describing the
- origin of the Work and reproducing the content of the NOTICE file.
-
- 7. Disclaimer of Warranty. Unless required by applicable law or
- agreed to in writing, Licensor provides the Work (and each
- Contributor provides its Contributions) on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
- implied, including, without limitation, any warranties or conditions
- of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
- PARTICULAR PURPOSE. You are solely responsible for determining the
- appropriateness of using or redistributing the Work and assume any
- risks associated with Your exercise of permissions under this License.
-
- 8. Limitation of Liability. In no event and under no legal theory,
- whether in tort (including negligence), contract, or otherwise,
- unless required by applicable law (such as deliberate and grossly
- negligent acts) or agreed to in writing, shall any Contributor be
- liable to You for damages, including any direct, indirect, special,
- incidental, or consequential damages of any character arising as a
- result of this License or out of the use or inability to use the
- Work (including but not limited to damages for loss of goodwill,
- work stoppage, computer failure or malfunction, or any and all
- other commercial damages or losses), even if such Contributor
- has been advised of the possibility of such damages.
-
- 9. Accepting Warranty or Additional Liability. While redistributing
- the Work or Derivative Works thereof, You may choose to offer,
- and charge a fee for, acceptance of support, warranty, indemnity,
- or other liability obligations and/or rights consistent with this
- License. However, in accepting such obligations, You may act only
- on Your own behalf and on Your sole responsibility, not on behalf
- of any other Contributor, and only if You agree to indemnify,
- defend, and hold each Contributor harmless for any liability
- incurred by, or claims asserted against, such Contributor by reason
- of your accepting any such warranty or additional liability.
-
- END OF TERMS AND CONDITIONS
-
- APPENDIX: How to apply the Apache License to your work.
-
- To apply the Apache License to your work, attach the following
- boilerplate notice, with the fields enclosed by brackets "[]"
- replaced with your own identifying information. (Don't include
- the brackets!) The text should be enclosed in the appropriate
- comment syntax for the file format. We also recommend that a
- file or class name and description of purpose be included on the
- same "printed page" as the copyright notice for easier
- identification within third-party archives.
-
- Copyright [yyyy] [name of copyright owner]
-
- Licensed 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.
diff --git a/modules/rocketmq/pom.xml b/modules/rocketmq/pom.xml
deleted file mode 100644
index 0d38a27..0000000
--- a/modules/rocketmq/pom.xml
+++ /dev/null
@@ -1,94 +0,0 @@
-<?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.
--->
-
-<!--
- POM file.
--->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <modelVersion>4.0.0</modelVersion>
-
- <parent>
- <groupId>org.apache.ignite</groupId>
- <artifactId>ignite-parent</artifactId>
- <version>1</version>
- <relativePath>../../parent</relativePath>
- </parent>
-
- <artifactId>ignite-rocketmq</artifactId>
- <version>2.9.0-SNAPSHOT</version>
- <url>http://ignite.apache.org</url>
-
- <dependencies>
- <dependency>
- <groupId>org.apache.ignite</groupId>
- <artifactId>ignite-core</artifactId>
- <version>${project.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.rocketmq</groupId>
- <artifactId>rocketmq-client</artifactId>
- <version>${rocketmq.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.rocketmq</groupId>
- <artifactId>rocketmq-namesrv</artifactId>
- <version>${rocketmq.version}</version>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.rocketmq</groupId>
- <artifactId>rocketmq-broker</artifactId>
- <version>${rocketmq.version}</version>
- <scope>test</scope>
- <exclusions>
- <exclusion>
- <groupId>io.netty</groupId>
- <artifactId>netty-tcnative</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>org.apache.ignite</groupId>
- <artifactId>ignite-spring</artifactId>
- <version>${project.version}</version>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.ignite</groupId>
- <artifactId>ignite-core</artifactId>
- <version>${project.version}</version>
- <type>test-jar</type>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.ignite</groupId>
- <artifactId>ignite-log4j</artifactId>
- <version>${project.version}</version>
- <scope>test</scope>
- </dependency>
- </dependencies>
-
-</project>
diff --git a/modules/rocketmq/src/main/java/org/apache/ignite/stream/rocketmq/RocketMQStreamer.java b/modules/rocketmq/src/main/java/org/apache/ignite/stream/rocketmq/RocketMQStreamer.java
deleted file mode 100644
index 67f1ce5..0000000
--- a/modules/rocketmq/src/main/java/org/apache/ignite/stream/rocketmq/RocketMQStreamer.java
+++ /dev/null
@@ -1,151 +0,0 @@
-/*
- * 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.
- */
-
-package org.apache.ignite.stream.rocketmq;
-
-import java.util.List;
-import org.apache.ignite.IgniteDataStreamer;
-import org.apache.ignite.IgniteException;
-import org.apache.ignite.IgniteLogger;
-import org.apache.ignite.internal.util.typedef.internal.A;
-import org.apache.ignite.stream.StreamAdapter;
-import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
-import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
-import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
-import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
-import org.apache.rocketmq.client.exception.MQClientException;
-import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
-import org.apache.rocketmq.common.message.MessageExt;
-
-/**
- * Streamer that subscribes to a RocketMQ topic amd feeds messages into {@link IgniteDataStreamer} instance.
- */
-public class RocketMQStreamer<K, V> extends StreamAdapter<List<MessageExt>, K, V> implements MessageListenerConcurrently {
- /** Logger. */
- private IgniteLogger log;
-
- /** RocketMQ consumer. */
- private DefaultMQPushConsumer consumer;
-
- /** State. */
- private volatile boolean stopped = true;
-
- /** Topic to subscribe to. */
- private String topic;
-
- /** Consumer group. */
- private String consumerGrp;
-
- /** Name server address. */
- private String nameSrvAddr;
-
- /**
- * Starts streamer.
- *
- * @throws IgniteException If failed.
- */
- public void start() {
- if (!stopped)
- throw new IgniteException("Attempted to start an already started RocketMQ streamer");
-
- // validate parameters.
- A.notNull(getStreamer(), "streamer");
- A.notNull(getIgnite(), "ignite");
- A.notNull(topic, "topic");
- A.notNull(consumerGrp, "consumer group");
- A.notNullOrEmpty(nameSrvAddr, "nameserver address");
- A.ensure(null != getMultipleTupleExtractor(), "Multiple tuple extractor must be configured");
-
- log = getIgnite().log();
-
- consumer = new DefaultMQPushConsumer(consumerGrp);
-
- consumer.setNamesrvAddr(nameSrvAddr);
- consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
-
- try {
- consumer.subscribe(topic, "*");
- }
- catch (MQClientException e) {
- throw new IgniteException("Failed to subscribe to " + topic, e);
- }
-
- consumer.registerMessageListener(this);
-
- try {
- consumer.start();
- }
- catch (MQClientException e) {
- throw new IgniteException("Failed to start the streamer", e);
- }
-
- stopped = false;
- }
-
- /**
- * Stops streamer.
- */
- public void stop() {
- if (consumer != null)
- consumer.shutdown();
-
- stopped = true;
- }
-
- /**
- * Implements {@link MessageListenerConcurrently#consumeMessage(List, ConsumeConcurrentlyContext)} to receive
- * messages.
- *
- * {@inheritDoc}
- */
- @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
- ConsumeConcurrentlyContext context) {
- if (log.isDebugEnabled())
- log.debug("Received " + msgs.size() + " messages");
-
- addMessage(msgs);
-
- return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
- }
-
- /**
- * Sets the topic to subscribe to.
- *
- * @param topic The topic to subscribe to.
- */
- public void setTopic(String topic) {
- this.topic = topic;
- }
-
- /**
- * Sets the name of the consumer group.
- *
- * @param consumerGrp Consumer group name.
- */
- public void setConsumerGrp(String consumerGrp) {
- this.consumerGrp = consumerGrp;
- }
-
- /**
- * Sets the name server address.
- *
- * @param nameSrvAddr Name server address
- */
- public void setNameSrvAddr(String nameSrvAddr) {
- this.nameSrvAddr = nameSrvAddr;
- }
-}
diff --git a/modules/rocketmq/src/main/java/org/apache/ignite/stream/rocketmq/package-info.java b/modules/rocketmq/src/main/java/org/apache/ignite/stream/rocketmq/package-info.java
deleted file mode 100644
index f743696..0000000
--- a/modules/rocketmq/src/main/java/org/apache/ignite/stream/rocketmq/package-info.java
+++ /dev/null
@@ -1,21 +0,0 @@
-/*
- * 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.
- */
-
-/**
- * Contains implementation of RocketMQStreamer tests.
- */
-package org.apache.ignite.stream.rocketmq;
diff --git a/modules/rocketmq/src/test/java/org/apache/ignite/stream/rocketmq/RocketMQStreamerTest.java b/modules/rocketmq/src/test/java/org/apache/ignite/stream/rocketmq/RocketMQStreamerTest.java
deleted file mode 100644
index 67b0f8a..0000000
--- a/modules/rocketmq/src/test/java/org/apache/ignite/stream/rocketmq/RocketMQStreamerTest.java
+++ /dev/null
@@ -1,215 +0,0 @@
-/*
- * 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.
- */
-
-package org.apache.ignite.stream.rocketmq;
-
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.UUID;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import org.apache.ignite.Ignite;
-import org.apache.ignite.IgniteCache;
-import org.apache.ignite.IgniteDataStreamer;
-import org.apache.ignite.cache.CachePeekMode;
-import org.apache.ignite.events.CacheEvent;
-import org.apache.ignite.internal.IgniteInterruptedCheckedException;
-import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.lang.IgniteBiPredicate;
-import org.apache.ignite.stream.StreamMultipleTupleExtractor;
-import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
-import org.apache.rocketmq.client.producer.DefaultMQProducer;
-import org.apache.rocketmq.common.TopicConfig;
-import org.apache.rocketmq.common.message.Message;
-import org.apache.rocketmq.common.message.MessageExt;
-import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
-import org.junit.Test;
-
-import static org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT;
-import static org.apache.ignite.stream.rocketmq.TestRocketMQServer.NAME_SERVER_PORT;
-import static org.apache.ignite.stream.rocketmq.TestRocketMQServer.TEST_IP;
-
-/**
- * Test for {@link RocketMQStreamer}.
- */
-public class RocketMQStreamerTest extends GridCommonAbstractTest {
- /** Test topic. */
- private static final String TOPIC_NAME = "testTopic";
-
- /** Test consumer group. */
- private static final String CONSUMER_GRP = "testConsumerGrp";
-
- /** Test server. */
- private static TestRocketMQServer testRocketMQServer;
-
- /** Number of events to handle. */
- private static final int EVT_NUM = 1000;
-
- /** {@inheritDoc} */
- @SuppressWarnings("unchecked")
- @Override protected void beforeTest() throws Exception {
- grid().<Integer, String>getOrCreateCache(defaultCacheConfiguration());
- }
-
- /** {@inheritDoc} */
- @Override protected void afterTest() throws Exception {
- grid().cache(DEFAULT_CACHE_NAME).clear();
- }
-
- /** {@inheritDoc} */
- @Override protected void beforeTestsStarted() throws Exception {
- testRocketMQServer = new TestRocketMQServer(log);
- }
-
- /** {@inheritDoc} */
- @Override protected void afterTestsStopped() throws Exception {
- if (testRocketMQServer != null)
- testRocketMQServer.shutdown();
- }
-
- /** Constructor. */
- public RocketMQStreamerTest() {
- super(true);
- }
-
- /**
- * Tests data is properly injected into the grid.
- *
- * @throws Exception If fails.
- */
- @Test
- public void testStreamer() throws Exception {
- RocketMQStreamer<String, byte[]> streamer = null;
-
- Ignite ignite = grid();
-
- try (IgniteDataStreamer<String, byte[]> dataStreamer = ignite.dataStreamer(DEFAULT_CACHE_NAME)) {
- dataStreamer.allowOverwrite(true);
- dataStreamer.autoFlushFrequency(10);
-
- streamer = new RocketMQStreamer<>();
-
- //configure.
- streamer.setIgnite(ignite);
- streamer.setStreamer(dataStreamer);
- streamer.setNameSrvAddr(TEST_IP + ":" + NAME_SERVER_PORT);
- streamer.setConsumerGrp(CONSUMER_GRP);
- streamer.setTopic(TOPIC_NAME);
- streamer.setMultipleTupleExtractor(new TestTupleExtractor());
-
- streamer.start();
-
- IgniteCache<String, String> cache = ignite.cache(DEFAULT_CACHE_NAME);
-
- assertEquals(0, cache.size(CachePeekMode.PRIMARY));
-
- final CountDownLatch latch = new CountDownLatch(EVT_NUM);
-
- IgniteBiPredicate<UUID, CacheEvent> putLsnr = new IgniteBiPredicate<UUID, CacheEvent>() {
- @Override public boolean apply(UUID uuid, CacheEvent evt) {
- assert evt != null;
-
- latch.countDown();
-
- return true;
- }
- };
-
- ignite.events(ignite.cluster().forCacheNodes(DEFAULT_CACHE_NAME)).remoteListen(putLsnr, null, EVT_CACHE_OBJECT_PUT);
-
- produceData();
-
- assertTrue(latch.await(30, TimeUnit.SECONDS));
-
- assertEquals(EVT_NUM, cache.size(CachePeekMode.PRIMARY));
- }
- finally {
- if (streamer != null)
- streamer.stop();
- }
- }
-
- /**
- * Test tuple extractor.
- */
- public static class TestTupleExtractor implements StreamMultipleTupleExtractor<List<MessageExt>, String, byte[]> {
-
- /** {@inheritDoc} */
- @Override public Map<String, byte[]> extract(List<MessageExt> msgs) {
- final Map<String, byte[]> map = new HashMap<>();
-
- for (MessageExt msg : msgs)
- map.put(msg.getMsgId(), msg.getBody());
-
- return map;
- }
- }
-
- /**
- * Adds data to RocketMQ.
- *
- * @throws Exception If fails.
- */
- private void produceData() throws Exception {
- initTopic(TOPIC_NAME, TEST_IP + ":" + NAME_SERVER_PORT);
-
- DefaultMQProducer producer = new DefaultMQProducer("testProducerGrp");
-
- producer.setNamesrvAddr(TEST_IP + ":" + NAME_SERVER_PORT);
-
- try {
- producer.start();
-
- for (int i = 0; i < EVT_NUM; i++)
- producer.send(new Message(TOPIC_NAME, "", String.valueOf(i).getBytes("UTF-8")));
- }
- catch (Exception e) {
- throw new Exception(e);
- }
- finally {
- producer.shutdown();
- }
- }
-
- /**
- * Initializes RocketMQ topic.
- *
- * @param topic Topic.
- * @param nsAddr Nameserver address.
- * @throws IgniteInterruptedCheckedException If fails.
- */
- private void initTopic(String topic, String nsAddr) throws Exception {
- DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt();
- defaultMQAdminExt.setNamesrvAddr(nsAddr);
- try {
- defaultMQAdminExt.start();
-
- TopicConfig topicConfig = new TopicConfig();
- topicConfig.setTopicName(topic);
- topicConfig.setReadQueueNums(4);
- topicConfig.setWriteQueueNums(4);
-
- defaultMQAdminExt.createAndUpdateTopicConfig(testRocketMQServer.getBrokerAddr(), topicConfig);
-
- U.sleep(100);
- }
- finally {
- defaultMQAdminExt.shutdown();
- }
- }
-}
diff --git a/modules/rocketmq/src/test/java/org/apache/ignite/stream/rocketmq/RocketMQStreamerTestSuite.java b/modules/rocketmq/src/test/java/org/apache/ignite/stream/rocketmq/RocketMQStreamerTestSuite.java
deleted file mode 100644
index c0d502b..0000000
--- a/modules/rocketmq/src/test/java/org/apache/ignite/stream/rocketmq/RocketMQStreamerTestSuite.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * 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.
- */
-
-package org.apache.ignite.stream.rocketmq;
-
-import org.junit.runner.RunWith;
-import org.junit.runners.Suite;
-
-/**
- * Apache RocketMQ streamers tests.
- */
-@RunWith(Suite.class)
-@Suite.SuiteClasses({
- RocketMQStreamerTest.class
-})
-public class RocketMQStreamerTestSuite {
-}
diff --git a/modules/rocketmq/src/test/java/org/apache/ignite/stream/rocketmq/TestRocketMQServer.java b/modules/rocketmq/src/test/java/org/apache/ignite/stream/rocketmq/TestRocketMQServer.java
deleted file mode 100644
index beece8e..0000000
--- a/modules/rocketmq/src/test/java/org/apache/ignite/stream/rocketmq/TestRocketMQServer.java
+++ /dev/null
@@ -1,148 +0,0 @@
-/*
- * 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.
- */
-
-package org.apache.ignite.stream.rocketmq;
-
-import java.util.UUID;
-import org.apache.ignite.IgniteLogger;
-import org.apache.rocketmq.broker.BrokerController;
-import org.apache.rocketmq.common.BrokerConfig;
-import org.apache.rocketmq.common.namesrv.NamesrvConfig;
-import org.apache.rocketmq.namesrv.NamesrvController;
-import org.apache.rocketmq.remoting.netty.NettyClientConfig;
-import org.apache.rocketmq.remoting.netty.NettyServerConfig;
-import org.apache.rocketmq.store.config.MessageStoreConfig;
-
-import static java.io.File.separator;
-
-/**
- * Test RocketMQ server handling a broker and a nameserver.
- */
-class TestRocketMQServer {
- /** Nameserver port. */
- protected static final int NAME_SERVER_PORT = 9000;
-
- /** Broker port. */
- private static final int BROKER_PORT = 8000;
-
- /** Broker HA port. */
- private static final int HA_PORT = 8001;
-
- /** Test ip address. */
- protected static final String TEST_IP = "127.0.0.1";
-
- /** Test broker name. */
- private static final String TEST_BROKER = "testBroker";
-
- /** Test cluster name. */
- private static final String TEST_CLUSTER = "testCluster";
-
- /** Nameserver. */
- private static NamesrvController nameSrv;
-
- /** Broker. */
- private static BrokerController broker;
-
- /** Logger. */
- private final IgniteLogger log;
-
- /**
- * Test server constructor.
- *
- * @param log Logger.
- */
- TestRocketMQServer(IgniteLogger log) {
- this.log = log;
-
- try {
- startNameServer();
- startBroker();
- }
- catch (Exception e) {
- throw new RuntimeException("Failed to start RocketMQ: " + e);
- }
- }
-
- /**
- * Starts a test nameserver.
- *
- * @throws Exception If fails.
- */
- private void startNameServer() throws Exception {
- NamesrvConfig namesrvConfig = new NamesrvConfig();
- NettyServerConfig nameServerNettyServerConfig = new NettyServerConfig();
-
- namesrvConfig.setKvConfigPath(System.getProperty("java.io.tmpdir") + separator + "namesrv" + separator + "kvConfig.json");
- nameServerNettyServerConfig.setListenPort(NAME_SERVER_PORT);
-
- nameSrv = new NamesrvController(namesrvConfig, nameServerNettyServerConfig);
-
- nameSrv.initialize();
- nameSrv.start();
-
- log.info("Started nameserver at " + NAME_SERVER_PORT);
- }
-
- /**
- * Starts a test broker.
- *
- * @throws Exception If fails.
- */
- private void startBroker() throws Exception {
- BrokerConfig brokerCfg = new BrokerConfig();
- NettyServerConfig nettySrvCfg = new NettyServerConfig();
- MessageStoreConfig storeCfg = new MessageStoreConfig();
-
- brokerCfg.setBrokerName(TEST_BROKER);
- brokerCfg.setBrokerClusterName(TEST_CLUSTER);
- brokerCfg.setBrokerIP1(TEST_IP);
- brokerCfg.setNamesrvAddr(TEST_IP + ":" + NAME_SERVER_PORT);
-
- storeCfg.setStorePathRootDir(System.getProperty("java.io.tmpdir") + separator + "store-" + UUID.randomUUID());
- storeCfg.setStorePathCommitLog(System.getProperty("java.io.tmpdir") + separator + "commitlog");
- storeCfg.setHaListenPort(HA_PORT);
-
- nettySrvCfg.setListenPort(BROKER_PORT);
-
- broker = new BrokerController(brokerCfg, nettySrvCfg, new NettyClientConfig(), storeCfg);
-
- broker.initialize();
- broker.start();
-
- log.info("Started broker [" + TEST_BROKER + "] at " + BROKER_PORT);
- }
-
- /**
- * Obtains the broker address.
- *
- * @return Broker address.
- */
- String getBrokerAddr() {
- return broker.getBrokerAddr();
- }
-
- /**
- * Shuts test server down.
- */
- void shutdown() {
- if (broker != null)
- broker.shutdown();
-
- if (nameSrv != null)
- nameSrv.shutdown();
- }
-}
diff --git a/modules/rocketmq/src/test/java/org/apache/ignite/stream/rocketmq/package-info.java b/modules/rocketmq/src/test/java/org/apache/ignite/stream/rocketmq/package-info.java
deleted file mode 100644
index eebf084..0000000
--- a/modules/rocketmq/src/test/java/org/apache/ignite/stream/rocketmq/package-info.java
+++ /dev/null
@@ -1,21 +0,0 @@
-/*
- * 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.
- */
-
-/**
- * Contains implementation of RocketMQStreamer.
- */
-package org.apache.ignite.stream.rocketmq;
diff --git a/parent/pom.xml b/parent/pom.xml
index 929e7f7..f301c117 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -114,7 +114,6 @@
<osgi.enterprise.version>5.0.0</osgi.enterprise.version>
<paho.version>1.0.2</paho.version>
<postgres.connector.version>42.2.5.jre7</postgres.connector.version>
- <rocketmq.version>4.4.0</rocketmq.version>
<scala210.jline.version>2.10.7</scala210.jline.version>
<scala210.library.version>2.10.7</scala210.library.version>
<scala.library.version>2.11.12</scala.library.version>
@@ -449,10 +448,6 @@
<packages>org.apache.ignite.springdata22.repository*</packages>
</group>
<group>
- <title>RocketMQ integration</title>
- <packages>org.apache.ignite.stream.rocketmq*</packages>
- </group>
- <group>
<title>Storm integration</title>
<packages>org.apache.ignite.stream.storm*</packages>
</group>
diff --git a/pom.xml b/pom.xml
index 325f383..ede52d1 100644
--- a/pom.xml
+++ b/pom.xml
@@ -88,7 +88,6 @@
<module>modules/web/ignite-websphere-test</module>
<module>modules/cassandra</module>
<module>modules/kubernetes</module>
- <module>modules/rocketmq</module>
<module>modules/sqlline</module>
<module>modules/ml</module>
<module>modules/ml/spark-model-parser</module>