This is an automated email from the ASF dual-hosted git repository.
mmerli 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 06f9e2d CLI for offload (#1865)
06f9e2d is described below
commit 06f9e2d126eba015bc9b0461014aa3b74794f546
Author: Ivan Kelly <[email protected]>
AuthorDate: Thu May 31 18:53:09 2018 +0200
CLI for offload (#1865)
* CLI for offload
This patch adds two CLI commands for offload, one to trigger offload,
and another to check offload status.
Triggering offload requires a size-threshold argument, which is the
maximum number of bytes which should be retained locally. The prefix
of ledgers over this threshold will be offloaded.
```
pulsar-admin topics offload --size-threshold 104857600
persistent://public/default/topic1
pulsar-admin topics offload-status -w persistent://public/default/topic1
```
Master Issue: #1511
* luc's review comments
---
.../org/apache/pulsar/admin/cli/CmdTopics.java | 94 +++++++++++++++++++++-
.../org/apache/pulsar/admin/cli/TestCmdTopics.java | 58 +++++++++++++
.../pulsar/tests/integration/TestS3Offload.java | 42 +++++-----
3 files changed, 175 insertions(+), 19 deletions(-)
diff --git
a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
index b8f93f7..4c14310 100644
---
a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
+++
b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
@@ -24,6 +24,7 @@ import static org.apache.commons.lang3.StringUtils.isNotBlank;
import com.beust.jcommander.Parameter;
import com.beust.jcommander.Parameters;
import com.beust.jcommander.converters.CommaParameterSplitter;
+import com.google.common.collect.Lists;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.JsonObject;
@@ -32,6 +33,7 @@ import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.Unpooled;
+import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
@@ -44,7 +46,7 @@ import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.impl.BatchMessageIdImpl;
import org.apache.pulsar.client.impl.MessageIdImpl;
-
+import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
@Parameters(commandDescription = "Operations on persistent topics")
public class CmdTopics extends CmdBase {
@@ -89,6 +91,8 @@ public class CmdTopics extends CmdBase {
jcommander.addCommand("terminate", new Terminate());
jcommander.addCommand("compact", new Compact());
jcommander.addCommand("compaction-status", new CompactionStatusCmd());
+ jcommander.addCommand("offload", new Offload());
+ jcommander.addCommand("offload-status", new OffloadStatusCmd());
}
@Parameters(commandDescription = "Get the list of topics under a
namespace.")
@@ -614,6 +618,94 @@ public class CmdTopics extends CmdBase {
}
}
+ static MessageId
findFirstLedgerWithinThreshold(List<PersistentTopicInternalStats.LedgerInfo>
ledgers,
+ long sizeThreshold) {
+ long suffixSize = 0L;
+
+ ledgers = Lists.reverse(ledgers);
+ for (PersistentTopicInternalStats.LedgerInfo l : ledgers) {
+ suffixSize += l.size;
+ if (suffixSize >= sizeThreshold) {
+ return new MessageIdImpl(l.ledgerId, 0L, -1);
+ }
+ }
+ return null;
+ }
+
+ @Parameters(commandDescription = "Trigger offload of data from a topic to
long-term storage (e.g. Amazon S3)")
+ private class Offload extends CliCommand {
+ @Parameter(names = { "-s", "--size-threshold" },
+ description = "Maximum amount of data to keep in BookKeeper
for the specified topic",
+ required = true)
+ private Long sizeThreshold;
+
+ @Parameter(description = "persistent://tenant/namespace/topic",
required = true)
+ private java.util.List<String> params;
+
+ @Override
+ void run() throws PulsarAdminException {
+ String persistentTopic = validatePersistentTopic(params);
+
+ PersistentTopicInternalStats stats =
topics.getInternalStats(persistentTopic);
+ if (stats.ledgers.size() < 1) {
+ throw new PulsarAdminException("Topic doesn't have any data");
+ }
+
+ LinkedList<PersistentTopicInternalStats.LedgerInfo> ledgers = new
LinkedList(stats.ledgers);
+ ledgers.get(ledgers.size()-1).size = stats.currentLedgerSize; //
doesn't get filled in now it seems
+ MessageId messageId = findFirstLedgerWithinThreshold(ledgers,
sizeThreshold);
+
+ if (messageId == null) {
+ System.out.println("Nothing to offload");
+ return;
+ }
+
+ topics.triggerOffload(persistentTopic, messageId);
+ System.out.println("Offload triggered for " + persistentTopic + "
for messages before " + messageId);
+ }
+ }
+
+ @Parameters(commandDescription = "Check the status of data offloading from
a topic to long-term storage")
+ private class OffloadStatusCmd extends CliCommand {
+ @Parameter(description = "persistent://tenant/namespace/topic",
required = true)
+ private java.util.List<String> params;
+
+ @Parameter(names = { "-w", "--wait-complete" },
+ description = "Wait for offloading to complete", required =
false)
+ private boolean wait = false;
+
+ @Override
+ void run() throws PulsarAdminException {
+ String persistentTopic = validatePersistentTopic(params);
+
+ try {
+ LongRunningProcessStatus status =
topics.offloadStatus(persistentTopic);
+ while (wait && status.status ==
LongRunningProcessStatus.Status.RUNNING) {
+ Thread.sleep(1000);
+ status = topics.offloadStatus(persistentTopic);
+ }
+
+ switch (status.status) {
+ case NOT_RUN:
+ System.out.println("Offload has not been run for " +
persistentTopic
+ + " since broker startup");
+ break;
+ case RUNNING:
+ System.out.println("Offload is currently running");
+ break;
+ case SUCCESS:
+ System.out.println("Offload was a success");
+ break;
+ case ERROR:
+ System.out.println("Error in offload");
+ throw new PulsarAdminException("Error offloading: " +
status.lastError);
+ }
+ } catch (InterruptedException e) {
+ throw new PulsarAdminException(e);
+ }
+ }
+ }
+
private static int validateTimeString(String s) {
char last = s.charAt(s.length() - 1);
String subStr = s.substring(0, s.length() - 1);
diff --git
a/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdTopics.java
b/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdTopics.java
new file mode 100644
index 0000000..df2d14c
--- /dev/null
+++
b/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdTopics.java
@@ -0,0 +1,58 @@
+/**
+ * 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.admin.cli;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.pulsar.client.impl.MessageIdImpl;
+import
org.apache.pulsar.common.policies.data.PersistentTopicInternalStats.LedgerInfo;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+public class TestCmdTopics {
+ private static LedgerInfo newLedger(long id, long entries, long size) {
+ LedgerInfo l = new LedgerInfo();
+ l.ledgerId = id;
+ l.entries = entries;
+ l.size = size;
+ return l;
+ }
+
+ @Test
+ public void testFindFirstLedgerWithinThreshold() throws Exception {
+ List<LedgerInfo> ledgers = new ArrayList<>();
+ ledgers.add(newLedger(0, 10, 1000));
+ ledgers.add(newLedger(1, 10, 2000));
+ ledgers.add(newLedger(2, 10, 3000));
+
+ // test huge threshold
+ Assert.assertNull(CmdTopics.findFirstLedgerWithinThreshold(ledgers,
Long.MAX_VALUE));
+
+ // test small threshold
+ Assert.assertEquals(CmdTopics.findFirstLedgerWithinThreshold(ledgers,
0),
+ new MessageIdImpl(2, 0, -1));
+
+ // test middling thresholds
+ Assert.assertEquals(CmdTopics.findFirstLedgerWithinThreshold(ledgers,
1000),
+ new MessageIdImpl(2, 0, -1));
+ Assert.assertEquals(CmdTopics.findFirstLedgerWithinThreshold(ledgers,
5000),
+ new MessageIdImpl(1, 0, -1));
+ }
+}
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
index c5b27f0..e3ee495 100644
---
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
@@ -93,7 +93,7 @@ public class TestS3Offload extends Arquillian {
}
@Test
- public void testPublishOffloadAndConsumeViaAdmin() throws Exception {
+ public void testPublishOffloadAndConsumeViaCLI() throws Exception {
PulsarClusterUtils.runOnAnyBroker(docker, CLUSTER_NAME,
"/pulsar/bin/pulsar-admin", "tenants",
"create", "--allowed-clusters", CLUSTER_NAME,
@@ -102,8 +102,8 @@ public class TestS3Offload extends Arquillian {
"/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 broker = PulsarClusterUtils.brokerSet(docker,
CLUSTER_NAME).stream().findAny().get();
+ String brokerIp = DockerUtils.getContainerIP(docker, broker);
String proxyIp = PulsarClusterUtils.proxySet(docker, CLUSTER_NAME)
.stream().map((c) -> DockerUtils.getContainerIP(docker,
c)).findFirst().get();
String serviceUrl = "pulsar://" + proxyIp + ":6650";
@@ -131,21 +131,27 @@ public class TestS3Offload extends Arquillian {
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");
- }
+ // first offload with a high threshold, nothing should offload
+ String output = DockerUtils.runCommand(docker, broker,
+ "/pulsar/bin/pulsar-admin", "topics",
+ "offload", "--size-threshold",
String.valueOf(Long.MAX_VALUE),
+ topic);
+ Assert.assertTrue(output.contains("Nothing to offload"));
+
+ output = DockerUtils.runCommand(docker, broker,
+ "/pulsar/bin/pulsar-admin", "topics", "offload-status",
topic);
+ Assert.assertTrue(output.contains("Offload has not been run"));
+
+ // offload with a low threshold
+ output = DockerUtils.runCommand(docker, broker,
+ "/pulsar/bin/pulsar-admin", "topics",
+ "offload", "--size-threshold", "0",
+ topic);
+ Assert.assertTrue(output.contains("Offload triggered"));
+
+ output = DockerUtils.runCommand(docker, broker,
+ "/pulsar/bin/pulsar-admin", "topics", "offload-status",
"-w", topic);
+ Assert.assertTrue(output.contains("Offload was a success"));
}
log.info("Kill ledger");
--
To stop receiving notification emails like this one, please contact
[email protected].
