refactor 1

pom.xml

@@ -2565,6 +2565,9 @@ flexible messaging model and an intuitive client API.</description>
         <module>pulsar-package-management</module>
         <!-- package management releated modules (end) -->
 
+        <module>pulsar-inttest-lib</module>
+        <module>pulsar-inttest-client</module>
+
         <!-- all these 3 modules should be put at the end in this exact sequence -->
         <module>distribution</module>
         <module>docker</module>
@@ -2621,6 +2624,9 @@ flexible messaging model and an intuitive client API.</description>
 
         <module>pulsar-client-messagecrypto-bc</module>
 
+        <module>pulsar-inttest-lib</module>
+        <module>pulsar-inttest-client</module>
+
         <!-- all these modules should be put at the end in this exact sequence -->
         <module>distribution</module>
         <module>pulsar-metadata</module>

pulsar-inttest-client/pom.xml

@@ -0,0 +1,100 @@
+<?xml version="1.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.
+
+-->
+<project
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
+  xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>org.apache.pulsar</groupId>
+    <artifactId>pulsar</artifactId>
+    <version>4.1.0-SNAPSHOT</version>
+  </parent>
+
+  <artifactId>pulsar-inttest-client</artifactId>
+  <description>Pulsar Integration Common Client Test Classes</description>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.pulsar</groupId>
+      <artifactId>pulsar-inttest-lib</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.assertj</groupId>
+      <artifactId>assertj-core</artifactId>
+      <version>${assertj-core.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.pulsar</groupId>
+      <artifactId>pulsar-client-api</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.pulsar</groupId>
+      <artifactId>pulsar-client-admin-api</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.awaitility</groupId>
+      <artifactId>awaitility</artifactId>
+      <version>${awaitility.version}</version>
+      <scope>compile</scope>
+    </dependency>
+  </dependencies>
+
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.gaul</groupId>
+        <artifactId>modernizer-maven-plugin</artifactId>
+        <configuration>
+          <failOnViolations>true</failOnViolations>
+          <javaVersion>8</javaVersion>
+        </configuration>
+        <executions>
+          <execution>
+            <id>modernizer</id>
+            <phase>verify</phase>
+            <goals>
+              <goal>modernizer</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
+
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-checkstyle-plugin</artifactId>
+        <executions>
+          <execution>
+            <id>checkstyle</id>
+            <phase>verify</phase>
+            <goals>
+              <goal>check</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
+
+    </plugins>
+  </build>
+</project>

pulsar-inttest-client/src/main/java/org/apache/pulsar/tests/integration/GeoRepIntegTest.java

@@ -0,0 +1,44 @@
+/*
+ * 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.pulsar.tests.integration;
+
+import java.io.IOException;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.PulsarClient;
+
+public class GeoRepIntegTest implements Cloneable {
+    public PulsarClient clientA;
+    public PulsarClient clientB;
+    public PulsarAdmin adminA;
+    public PulsarAdmin adminB;
+
+    public GeoRepIntegTest(PulsarClient clientA, PulsarAdmin adminA, PulsarClient clientB, PulsarAdmin adminB) {
+        this.clientA = clientA;
+        this.adminA = adminA;
+        this.clientB = clientB;
+        this.adminB = adminB;
+    }
+
+    public void close() throws IOException {
+        clientA.close();
+        clientB.close();
+        adminA.close();
+        adminB.close();
+    }
+}

pulsar-inttest-client/src/main/java/org/apache/pulsar/tests/integration/IntegTest.java

@@ -0,0 +1,38 @@
+/*
+ * 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.pulsar.tests.integration;
+
+import java.io.IOException;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.PulsarClient;
+
+public class IntegTest implements Cloneable {
+    public PulsarClient client;
+    public PulsarAdmin admin;
+
+    public IntegTest(PulsarClient client, PulsarAdmin admin) {
+        this.client = client;
+        this.admin = admin;
+    }
+
+    public void close() throws IOException {
+        client.close();
+        admin.close();
+    }
+}

pulsar-inttest-client/src/main/java/org/apache/pulsar/tests/integration/messaging/DelayMessaging.java

@@ -0,0 +1,105 @@
+/*
+ * 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.pulsar.tests.integration.messaging;
+
+import static org.apache.pulsar.tests.integration.utils.IntegTestUtils.getPartitionedTopic;
+import static org.assertj.core.api.Assertions.assertThat;
+import java.util.concurrent.TimeUnit;
+import lombok.Cleanup;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.DeadLetterPolicy;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.tests.integration.IntegTest;
+
+/**
+ * Delay messaging test.
+ */
+@Slf4j
+public class DelayMessaging extends IntegTest {
+
+    public DelayMessaging(PulsarClient client, PulsarAdmin admin) {
+        super(client, admin);
+    }
+
+    public void delayMsgBlockTest() throws Exception {
+
+        String topic = getPartitionedTopic(admin, "testDelayMsgBlock", true, 3);
+
+        String retryTopic = topic + "-RETRY";
+        String deadLetterTopic = topic + "-DLT";
+
+        @Cleanup
+        Producer<byte[]> producer = client.newProducer()
+                .topic(topic)
+                .create();
+
+        final int redeliverCnt = 10;
+        final int delayTimeSeconds = 5;
+        @Cleanup
+        Consumer<byte[]> consumer = client.newConsumer()
+                .topic(topic)
+                .subscriptionName("test")
+                .subscriptionType(SubscriptionType.Shared)
+                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+                .enableRetry(true)
+                .deadLetterPolicy(DeadLetterPolicy.builder()
+                        .maxRedeliverCount(redeliverCnt)
+                        .retryLetterTopic(retryTopic)
+                        .deadLetterTopic(deadLetterTopic)
+                        .build())
+                .receiverQueueSize(100)
+                .ackTimeout(60, TimeUnit.SECONDS)
+                .subscribe();
+
+        producer.newMessage().value("hello".getBytes()).send();
+
+        // receive message at first time
+        Message<byte[]> message = consumer.receive(delayTimeSeconds * 2, TimeUnit.SECONDS);
+        assertThat(message).isNotNull().as("Can't receive message at the first time.");
+        consumer.reconsumeLater(message, delayTimeSeconds, TimeUnit.SECONDS);
+
+        // receive retry messages
+        for (int i = 0; i < redeliverCnt; i++) {
+            message = consumer.receive(delayTimeSeconds * 2, TimeUnit.SECONDS);
+            assertThat(message)
+                    .isNotNull()
+                    .as("Consumer can't receive message in double delayTimeSeconds time "
+                            + delayTimeSeconds * 2 + "s");
+            log.info("receive msg. reConsumeTimes: {}", message.getProperty("RECONSUMETIMES"));
+            consumer.reconsumeLater(message, delayTimeSeconds, TimeUnit.SECONDS);
+        }
+
+        @Cleanup
+        Consumer<byte[]> dltConsumer = client.newConsumer()
+                .topic(deadLetterTopic)
+                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+                .subscriptionName("test")
+                .subscribe();
+
+        message = dltConsumer.receive(10, TimeUnit.SECONDS);
+        assertThat(message).isNotNull().as("Dead letter topic consumer can't receive message.");
+    }
+
+}

pulsar-inttest-client/src/main/java/org/apache/pulsar/tests/integration/messaging/GeoReplication.java

@@ -0,0 +1,84 @@
+/*
+ * 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.pulsar.tests.integration.messaging;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
+import java.nio.charset.StandardCharsets;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import lombok.Cleanup;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.tests.integration.GeoRepIntegTest;
+import org.awaitility.Awaitility;
+
+/**
+ * Geo replication test.
+ */
+@Slf4j
+public class GeoReplication extends GeoRepIntegTest {
+
+    public GeoReplication(PulsarClient clientA, PulsarAdmin adminA, PulsarClient clientB, PulsarAdmin adminB) {
+        super(clientA, adminA, clientB, adminB);
+    }
+
+    public void testTopicReplication(String domain) throws Exception {
+
+        String topic = domain + "://public/default/testTopicReplication-" + UUID.randomUUID();
+        Awaitility.await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+            try {
+                adminA.topics().createPartitionedTopic(topic, 10);
+            } catch (Exception e) {
+                log.error("Failed to create partitioned topic {}.", topic, e);
+                fail("Failed to create partitioned topic " + topic);
+            }
+            assertThat(adminA.topics().getPartitionedTopicMetadata(topic).partitions).isEqualTo(10);
+        });
+        log.info("Test geo-replication produce and consume for topic {}.", topic);
+
+        @Cleanup
+        Producer<byte[]> p = clientA.newProducer()
+                .topic(topic)
+                .create();
+        log.info("Successfully create producer in cluster {} for topic {}.", "cluster1", topic);
+
+        @Cleanup
+        Consumer<byte[]> c = clientB.newConsumer()
+                .topic(topic)
+                .subscriptionName("geo-sub")
+                .subscribe();
+        log.info("Successfully create consumer in cluster {} for topic {}.", "cluster2", topic);
+
+        for (int i = 0; i < 10; i++) {
+            p.send(String.format("Message [%d]", i).getBytes(StandardCharsets.UTF_8));
+        }
+        log.info("Successfully produce message to cluster {} for topic {}.", "cluster1", topic);
+
+        for (int i = 0; i < 10; i++) {
+            Message<byte[]> message = c.receive(10, TimeUnit.SECONDS);
+            assertThat(message).isNotNull();
+        }
+        log.info("Successfully consume message from cluster {} for topic {}.", "cluster2", topic);
+    }
+}