This is an automated email from the ASF dual-hosted git repository.
sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new fdb35d4 Integration test for s3offload (#1829)
fdb35d4 is described below
commit fdb35d449ceae8ec9859bdbf077fd1bbe87d5bad
Author: Ivan Kelly <[email protected]>
AuthorDate: Wed May 23 19:11:22 2018 +0100
Integration test for s3offload (#1829)
Starts a cluster with s3offload enabled. Write enough entries to roll
the ledger. Offload. Manually delete the first ledger. Reboot cluster
to clear caches. Then read back the whole topic.
Master issue: #1511
---
...luster-2-bookie-1-broker-unstarted-with-s3.yaml | 156 +++++++++++++++++++
tests/integration/pom.xml | 1 +
tests/integration/s3-offload/pom.xml | 57 +++++++
.../pulsar/tests/integration/TestS3Offload.java | 173 +++++++++++++++++++++
.../src/test/resources/arquillian.xml} | 30 ++--
5 files changed, 398 insertions(+), 19 deletions(-)
diff --git
a/tests/integration-tests-topologies/src/main/resources/cube-definitions/single-cluster-2-bookie-1-broker-unstarted-with-s3.yaml
b/tests/integration-tests-topologies/src/main/resources/cube-definitions/single-cluster-2-bookie-1-broker-unstarted-with-s3.yaml
new file mode 100644
index 0000000..b80ef7c
--- /dev/null
+++
b/tests/integration-tests-topologies/src/main/resources/cube-definitions/single-cluster-2-bookie-1-broker-unstarted-with-s3.yaml
@@ -0,0 +1,156 @@
+#
+# 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.
+#
+
+networks:
+ pulsarnet*:
+ driver: bridge
+
+zookeeper*:
+ image: apachepulsar/pulsar-test-latest-version:latest
+ await:
+ strategy: org.apache.pulsar.tests.NoopAwaitStrategy
+ env: [ZOOKEEPER_SERVERS=zookeeper]
+ labels:
+ cluster: test
+ service: zookeeper
+ entryPoint: [bin/run-local-zk.sh]
+ aliases:
+ - zookeeper
+ beforeStop:
+ - customBeforeStopAction:
+ strategy: org.apache.pulsar.tests.PulsarLogsToTargetDirStopAction
+ networkMode: pulsarnet*
+
+configuration-store*:
+ image: apachepulsar/pulsar-test-latest-version:latest
+ await:
+ strategy: org.apache.pulsar.tests.NoopAwaitStrategy
+ env: [ZOOKEEPER_SERVERS=configuration-store]
+ labels:
+ cluster: test
+ service: configuration-store
+ entryPoint: [bin/run-global-zk.sh]
+ aliases:
+ - configuration-store
+ beforeStop:
+ - customBeforeStopAction:
+ strategy: org.apache.pulsar.tests.PulsarLogsToTargetDirStopAction
+ networkMode: pulsarnet*
+
+init*:
+ image: apachepulsar/pulsar-test-latest-version:latest
+ await:
+ strategy: org.apache.pulsar.tests.NoopAwaitStrategy
+ env:
+ - cluster=test
+ - zkServers=zookeeper
+ - configurationStore=configuration-store:2184
+ - pulsarNode=pulsar-broker1
+ labels:
+ cluster: test
+ service: init
+ entryPoint: [bin/init-cluster.sh]
+ beforeStop:
+ - customBeforeStopAction:
+ strategy: org.apache.pulsar.tests.LogToTargetDirStopAction
+ networkMode: pulsarnet*
+
+bookkeeper1*:
+ image: apachepulsar/pulsar-test-latest-version:latest
+ await:
+ strategy: org.apache.pulsar.tests.NoopAwaitStrategy
+ env:
+ - zkServers=zookeeper
+ - useHostNameAsBookieID=true
+ labels:
+ cluster: test
+ service: bookie
+ entryPoint: [bin/run-bookie.sh]
+ beforeStop:
+ - customBeforeStopAction:
+ strategy: org.apache.pulsar.tests.PulsarLogsToTargetDirStopAction
+ networkMode: pulsarnet*
+
+bookkeeper2*:
+ image: apachepulsar/pulsar-test-latest-version:latest
+ await:
+ strategy: org.apache.pulsar.tests.NoopAwaitStrategy
+ env:
+ - zkServers=zookeeper
+ - useHostNameAsBookieID=true
+ labels:
+ cluster: test
+ service: bookie
+ entryPoint: [bin/run-bookie.sh]
+ beforeStop:
+ - customBeforeStopAction:
+ strategy: org.apache.pulsar.tests.PulsarLogsToTargetDirStopAction
+ networkMode: pulsarnet*
+
+pulsar-broker1*:
+ image: apachepulsar/pulsar-test-latest-version:latest
+ await:
+ strategy: org.apache.pulsar.tests.NoopAwaitStrategy
+ aliases:
+ - pulsar-broker1
+ env:
+ - zookeeperServers=zookeeper
+ - configurationStoreServers=configuration-store:2184
+ - clusterName=test
+ - NO_AUTOSTART=true
+ labels:
+ cluster: test
+ service: pulsar-broker
+ entryPoint: [bin/run-broker.sh]
+ beforeStop:
+ - customBeforeStopAction:
+ strategy: org.apache.pulsar.tests.PulsarLogsToTargetDirStopAction
+ networkMode: pulsarnet*
+
+pulsar-proxy*:
+ image: apachepulsar/pulsar-test-latest-version:latest
+ await:
+ strategy: org.apache.pulsar.tests.NoopAwaitStrategy
+ env:
+ - zookeeperServers=zookeeper
+ - configurationStoreServers=configuration-store:2184
+ - clusterName=test
+ - NO_AUTOSTART=true
+ labels:
+ cluster: test
+ service: pulsar-proxy
+ entryPoint: [bin/run-proxy.sh]
+ beforeStop:
+ - customBeforeStopAction:
+ strategy: org.apache.pulsar.tests.PulsarLogsToTargetDirStopAction
+ networkMode: pulsarnet*
+
+s3*:
+ image: adobe/s3mock
+ await:
+ strategy: org.apache.pulsar.tests.NoopAwaitStrategy
+ env:
+ - initialBuckets=pulsar-integtest
+ labels:
+ cluster: test
+ service: s3
+ beforeStop:
+ - customBeforeStopAction:
+ strategy: org.apache.pulsar.tests.LogToTargetDirStopAction
+ networkMode: pulsarnet*
diff --git a/tests/integration/pom.xml b/tests/integration/pom.xml
index 1245571..1dd3ad3 100644
--- a/tests/integration/pom.xml
+++ b/tests/integration/pom.xml
@@ -36,5 +36,6 @@
<module>smoke</module>
<module>compaction</module>
<module>cli</module>
+ <module>s3-offload</module>
</modules>
</project>
diff --git a/tests/integration/s3-offload/pom.xml
b/tests/integration/s3-offload/pom.xml
new file mode 100644
index 0000000..2d4812a
--- /dev/null
+++ b/tests/integration/s3-offload/pom.xml
@@ -0,0 +1,57 @@
+<?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.
+
+-->
+<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.pulsar.tests</groupId>
+ <artifactId>integration-tests-base</artifactId>
+ <version>2.1.0-incubating-SNAPSHOT</version>
+ <relativePath>../../integration-tests-base</relativePath>
+ </parent>
+
+ <groupId>org.apache.pulsar.tests.integration</groupId>
+ <artifactId>s3-offload</artifactId>
+ <packaging>jar</packaging>
+ <name>Apache Pulsar :: Tests :: Integration Tests :: S3 Offload</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar-client</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar-client-admin-original</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>managed-ledger</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ </dependencies>
+</project>
diff --git
a/tests/integration/s3-offload/src/test/java/org/apache/pulsar/tests/integration/TestS3Offload.java
b/tests/integration/s3-offload/src/test/java/org/apache/pulsar/tests/integration/TestS3Offload.java
new file mode 100644
index 0000000..c5b27f0
--- /dev/null
+++
b/tests/integration/s3-offload/src/test/java/org/apache/pulsar/tests/integration/TestS3Offload.java
@@ -0,0 +1,173 @@
+/**
+ * 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 com.github.dockerjava.api.DockerClient;
+import com.google.common.collect.ImmutableMap;
+
+import java.net.URL;
+
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+
+import org.apache.bookkeeper.mledger.ManagedLedgerFactory;
+import org.apache.bookkeeper.mledger.ManagedLedgerInfo;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl;
+
+import org.apache.pulsar.client.admin.LongRunningProcessStatus;
+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.MessageId;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+
+import org.apache.pulsar.tests.DockerUtils;
+import org.apache.pulsar.tests.PulsarClusterUtils;
+import org.jboss.arquillian.test.api.ArquillianResource;
+import org.jboss.arquillian.testng.Arquillian;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+public class TestS3Offload extends Arquillian {
+ private static final Logger log =
LoggerFactory.getLogger(TestS3Offload.class);
+
+ private static final String CLUSTER_NAME = "test";
+ private static final int ENTRIES_PER_LEDGER = 10;
+
+ @ArquillianResource
+ DockerClient docker;
+
+ @BeforeMethod
+ public void configureAndStartBrokers() throws Exception {
+
+ String s3ip = DockerUtils.cubeIdsWithLabels(
+ docker, ImmutableMap.of("service", "s3", "cluster",
CLUSTER_NAME))
+ .stream().map((c) -> DockerUtils.getContainerIP(docker,
c)).findFirst().get();
+
+ String brokerConfFile = "/pulsar/conf/broker.conf";
+ for (String b : PulsarClusterUtils.brokerSet(docker, CLUSTER_NAME)) {
+ PulsarClusterUtils.updateConf(docker, b, brokerConfFile,
+ "managedLedgerMaxEntriesPerLedger",
String.valueOf(ENTRIES_PER_LEDGER));
+ PulsarClusterUtils.updateConf(docker, b, brokerConfFile,
+ "managedLedgerMinLedgerRolloverTimeMinutes", "0");
+ PulsarClusterUtils.updateConf(docker, b, brokerConfFile,
+ "managedLedgerOffloadDriver", "s3");
+ PulsarClusterUtils.updateConf(docker, b, brokerConfFile,
+ "s3ManagedLedgerOffloadBucket", "pulsar-integtest");
+ PulsarClusterUtils.updateConf(docker, b, brokerConfFile,
+ "s3ManagedLedgerOffloadServiceEndpoint", "http://"; + s3ip
+ ":9090");
+ }
+
+ Assert.assertTrue(PulsarClusterUtils.startAllBrokers(docker,
CLUSTER_NAME));
+ Assert.assertTrue(PulsarClusterUtils.startAllProxies(docker,
CLUSTER_NAME));
+ }
+
+ @AfterMethod
+ public void teardownBrokers() throws Exception {
+ PulsarClusterUtils.stopAllProxies(docker, CLUSTER_NAME);
+ Assert.assertTrue(PulsarClusterUtils.stopAllBrokers(docker,
CLUSTER_NAME));
+
+ }
+
+ @Test
+ public void testPublishOffloadAndConsumeViaAdmin() throws Exception {
+ PulsarClusterUtils.runOnAnyBroker(docker, CLUSTER_NAME,
+ "/pulsar/bin/pulsar-admin", "tenants",
+ "create", "--allowed-clusters", CLUSTER_NAME,
+ "--admin-roles", "offload-admin", "s3-offload-test");
+ PulsarClusterUtils.runOnAnyBroker(docker, CLUSTER_NAME,
+ "/pulsar/bin/pulsar-admin", "namespaces",
+ "create", "--clusters", CLUSTER_NAME, "s3-offload-test/ns1");
+
+ String brokerIp = PulsarClusterUtils.brokerSet(docker, CLUSTER_NAME)
+ .stream().map((c) -> DockerUtils.getContainerIP(docker,
c)).findFirst().get();
+ String proxyIp = PulsarClusterUtils.proxySet(docker, CLUSTER_NAME)
+ .stream().map((c) -> DockerUtils.getContainerIP(docker,
c)).findFirst().get();
+ String serviceUrl = "pulsar://" + proxyIp + ":6650";
+ String adminUrl = "http://"; + brokerIp + ":8080";
+ String topic = "persistent://s3-offload-test/ns1/topic1";
+
+ ClientConfiguration bkConf = new ClientConfiguration();
+ bkConf.setZkServers(PulsarClusterUtils.zookeeperConnectString(docker,
CLUSTER_NAME));
+
+ long firstLedger = -1;
+ try(PulsarClient client = PulsarClient.create(serviceUrl);
+ Producer producer = client.createProducer(topic)) {
+ client.subscribe(topic, "my-sub").close();
+ // write enough to topic to make it roll
+ int i = 0;
+ for (; i < ENTRIES_PER_LEDGER*1.5; i++) {
+ producer.send(("offload-message"+i).getBytes());
+ }
+ MessageId latestMessage =
producer.send(("offload-message"+i).getBytes());
+
+ // read managed ledger info, check ledgers exist
+ ManagedLedgerFactory mlf = new ManagedLedgerFactoryImpl(bkConf);
+ ManagedLedgerInfo info =
mlf.getManagedLedgerInfo("s3-offload-test/ns1/persistent/topic1");
+ Assert.assertEquals(info.ledgers.size(), 2);
+
+ firstLedger = info.ledgers.get(0).ledgerId;
+
+ // trigger offload
+ try (PulsarAdmin admin = new PulsarAdmin(new URL(adminUrl), "",
"")) {
+ log.info("Trigger offload");
+
+ admin.topics().triggerOffload(topic, latestMessage);
+
+ LongRunningProcessStatus status =
admin.topics().offloadStatus(topic);
+ while (status.status ==
LongRunningProcessStatus.Status.RUNNING) {
+ Thread.sleep(100);
+ status = admin.topics().offloadStatus(topic);
+ }
+ Assert.assertEquals(status.status,
LongRunningProcessStatus.Status.SUCCESS);
+
+ log.info("Offload complete");
+ }
+ }
+
+ log.info("Kill ledger");
+ // stop brokers to clear all caches, open handles, etc
+ Assert.assertTrue(PulsarClusterUtils.stopAllBrokers(docker,
CLUSTER_NAME));
+
+ // delete the first ledger, so that we cannot possibly read from it
+ try (BookKeeper bk = new BookKeeper(bkConf)) {
+ bk.deleteLedger(firstLedger);
+ }
+
+ // start all brokers again
+ Assert.assertTrue(PulsarClusterUtils.startAllBrokers(docker,
CLUSTER_NAME));
+
+ log.info("Read back the data (which would be in that first ledger)");
+ try(PulsarClient client = PulsarClient.create(serviceUrl);
+ Consumer consumer = client.subscribe(topic, "my-sub")) {
+ // read back from topic
+ for (int i = 0; i < ENTRIES_PER_LEDGER*1.5; i++) {
+ Message m = consumer.receive();
+ Assert.assertEquals("offload-message"+i, new
String(m.getData()));
+ }
+ }
+ }
+}
diff --git a/tests/integration/pom.xml
b/tests/integration/s3-offload/src/test/resources/arquillian.xml
similarity index 52%
copy from tests/integration/pom.xml
copy to tests/integration/s3-offload/src/test/resources/arquillian.xml
index 1245571..14714d2 100644
--- a/tests/integration/pom.xml
+++ b/tests/integration/s3-offload/src/test/resources/arquillian.xml
@@ -1,4 +1,4 @@
-<?xml version="1.0" encoding="UTF-8"?>
+<?xml version="1.0"?>
<!--
Licensed to the Apache Software Foundation (ASF) under one
@@ -19,22 +19,14 @@
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";>
- <packaging>pom</packaging>
- <modelVersion>4.0.0</modelVersion>
- <parent>
- <groupId>org.apache.pulsar.tests</groupId>
- <artifactId>tests-parent</artifactId>
- <version>2.1.0-incubating-SNAPSHOT</version>
- </parent>
+<arquillian xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xmlns="http://jboss.org/schema/arquillian";
+ xsi:schemaLocation="http://jboss.org/schema/arquillian
+
http://jboss.org/schema/arquillian/arquillian_1_0.xsd";>
- <groupId>org.apache.pulsar.tests</groupId>
- <artifactId>integration</artifactId>
- <name>Apache Pulsar :: Tests :: Integration</name>
- <modules>
- <module>smoke</module>
- <module>compaction</module>
- <module>cli</module>
- </modules>
-</project>
+ <extension qualifier="docker">
+ <property name="definitionFormat">CUBE</property>
+ <property
name="dockerContainersResource">cube-definitions/single-cluster-2-bookie-1-broker-unstarted-with-s3.yaml</property>
+ </extension>
+
+</arquillian>
--
To stop receiving notification emails like this one, please contact
[email protected].
