This is an automated email from the ASF dual-hosted git repository.
zhaocong pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.9 by this push:
new a7c83f381bf [cherry-pick][branch-2.9] Fix
`validatePersistencePolicies` that Namespace/Topic persistent policies cannot
set to < 0 (#19687)
a7c83f381bf is described below
commit a7c83f381bf306aa888aca856d119b9bb012912a
Author: 道君 <[email protected]>
AuthorDate: Fri Mar 3 16:17:04 2023 +0800
[cherry-pick][branch-2.9] Fix `validatePersistencePolicies` that
Namespace/Topic persistent policies cannot set to < 0 (#19687)
Cherry-pick #18999
---
.../apache/pulsar/broker/admin/AdminResource.java | 15 +++++++------
.../apache/pulsar/broker/admin/NamespacesTest.java | 13 +++++++++++
.../pulsar/broker/admin/TopicPoliciesTest.java | 16 ++++++++++++++
.../broker/auth/MockedPulsarServiceBaseTest.java | 13 +++++++++++
.../org/apache/pulsar/admin/cli/CmdNamespaces.java | 25 +++++++++++++++-------
.../org/apache/pulsar/admin/cli/CmdTopics.java | 24 ++++++++++++++-------
6 files changed, 84 insertions(+), 22 deletions(-)
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
index e78b57aa8c9..1deac94b1b8 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java
@@ -811,12 +811,15 @@ public abstract class AdminResource extends
PulsarWebResource {
protected void validatePersistencePolicies(PersistencePolicies
persistence) {
checkNotNull(persistence, "persistence policies should not be null");
final ServiceConfiguration config = pulsar().getConfiguration();
- checkArgument(persistence.getBookkeeperEnsemble() <=
config.getManagedLedgerMaxEnsembleSize(),
- "Bookkeeper-Ensemble must be <= " +
config.getManagedLedgerMaxEnsembleSize());
- checkArgument(persistence.getBookkeeperWriteQuorum() <=
config.getManagedLedgerMaxWriteQuorum(),
- "Bookkeeper-WriteQuorum must be <= " +
config.getManagedLedgerMaxWriteQuorum());
- checkArgument(persistence.getBookkeeperAckQuorum() <=
config.getManagedLedgerMaxAckQuorum(),
- "Bookkeeper-AckQuorum must be <= " +
config.getManagedLedgerMaxAckQuorum());
+ checkArgument(persistence.getBookkeeperEnsemble() <=
config.getManagedLedgerMaxEnsembleSize()
+ && persistence.getBookkeeperEnsemble() > 0,
+ "Bookkeeper-Ensemble must be <= " +
config.getManagedLedgerMaxEnsembleSize() + " and > 0.");
+ checkArgument(persistence.getBookkeeperWriteQuorum() <=
config.getManagedLedgerMaxWriteQuorum()
+ && persistence.getBookkeeperWriteQuorum() > 0,
+ "Bookkeeper-WriteQuorum must be <= " +
config.getManagedLedgerMaxWriteQuorum() + " and > 0.");
+ checkArgument(persistence.getBookkeeperAckQuorum() <=
config.getManagedLedgerMaxAckQuorum()
+ && persistence.getBookkeeperAckQuorum() > 0,
+ "Bookkeeper-AckQuorum must be <= " +
config.getManagedLedgerMaxAckQuorum() + " and > 0.");
checkArgument(
(persistence.getBookkeeperEnsemble() >=
persistence.getBookkeeperWriteQuorum())
&& (persistence.getBookkeeperWriteQuorum() >=
persistence.getBookkeeperAckQuorum()),
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java
index 55441d9529f..6832a896256 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java
@@ -1072,6 +1072,19 @@ public class NamespacesTest extends
MockedPulsarServiceBaseTest {
assertEquals(persistence2, persistence1);
}
+ @Test(dataProvider = "invalidPersistentPolicies")
+ public void testSetIncorrectPersistentPolicies(int ensembleSize, int
writeQuorum, int ackQuorum) throws Exception {
+ NamespaceName testNs = this.testLocalNamespaces.get(0);
+ PersistencePolicies persistence1 = new
PersistencePolicies(ensembleSize, writeQuorum, ackQuorum, 0.0);
+ try {
+ namespaces.setPersistence(testNs.getTenant(), testNs.getCluster(),
testNs.getLocalName(), persistence1);
+ } catch (Exception ex) {
+ assertTrue(ex instanceof RestException);
+ RestException rex = (RestException) ex;
+ Assert.assertEquals(rex.getResponse().getStatus(),
Status.BAD_REQUEST.getStatusCode());
+ }
+ }
+
@Test
public void testPersistenceUnauthorized() throws Exception {
try {
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
index f82d7c01230..e6ef3f7f788 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java
@@ -832,6 +832,22 @@ public class TopicPoliciesTest extends
MockedPulsarServiceBaseTest {
consumer.close();
}
+ @Test(dataProvider = "invalidPersistentPolicies")
+ public void testSetIncorrectPersistentPolicies(int ensembleSize, int
writeQuorum, int ackQuorum) throws Exception {
+ admin.topics().createNonPartitionedTopic(persistenceTopic);
+ PersistencePolicies persistence1 = new
PersistencePolicies(ensembleSize, writeQuorum, ackQuorum, 0.0);
+
+ boolean failed = false;
+ try {
+ admin.topicPolicies().setPersistence(persistenceTopic,
persistence1);
+ } catch (PulsarAdminException e) {
+ failed = true;
+ Assert.assertEquals(e.getStatusCode(), 400);
+ }
+ assertTrue(failed);
+ admin.topics().delete(persistenceTopic);
+ }
+
@Test
public void testGetDispatchRateApplied() throws Exception {
final String topic = testTopic + UUID.randomUUID();
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
index 27ac9cb869d..374b55c2ee2 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
@@ -75,6 +75,7 @@ import org.apache.zookeeper.data.ACL;
import org.powermock.core.classloader.annotations.PowerMockIgnore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.testng.annotations.DataProvider;
/**
* Base class for all tests that need a Pulsar instance without a ZK and BK
cluster.
@@ -519,5 +520,17 @@ public abstract class MockedPulsarServiceBaseTest extends
TestRetrySupport {
}
}
+ @DataProvider(name = "invalidPersistentPolicies")
+ public Object[][] incorrectPersistentPolicies() {
+ return new Object[][] {
+ {0, 0, 0},
+ {1, 0, 0},
+ {0, 0, 1},
+ {0, 1, 0},
+ {1, 1, 0},
+ {1, 0, 1}
+ };
+ }
+
private static final Logger log =
LoggerFactory.getLogger(MockedPulsarServiceBaseTest.class);
}
diff --git
a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaces.java
b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaces.java
index f9f94c379e9..a9970c38a54 100644
---
a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaces.java
+++
b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaces.java
@@ -1211,24 +1211,33 @@ public class CmdNamespaces extends CmdBase {
private java.util.List<String> params;
@Parameter(names = { "-e",
- "--bookkeeper-ensemble" }, description = "Number of bookies to
use for a topic", required = true)
- private int bookkeeperEnsemble;
+ "--bookkeeper-ensemble" }, description = "Number of bookies to
use for a topic")
+ private int bookkeeperEnsemble = 2;
@Parameter(names = { "-w",
- "--bookkeeper-write-quorum" }, description = "How many writes
to make of each entry", required = true)
- private int bookkeeperWriteQuorum;
+ "--bookkeeper-write-quorum" }, description = "How many writes
to make of each entry")
+ private int bookkeeperWriteQuorum = 2;
@Parameter(names = { "-a",
- "--bookkeeper-ack-quorum" }, description = "Number of acks
(guaranteed copies) to wait for each entry", required = true)
- private int bookkeeperAckQuorum;
+ "--bookkeeper-ack-quorum" },
+ description = "Number of acks (guaranteed copies) to wait for
each entry")
+ private int bookkeeperAckQuorum = 2;
@Parameter(names = { "-r",
- "--ml-mark-delete-max-rate" }, description = "Throttling rate
of mark-delete operation (0 means no throttle)", required = true)
- private double managedLedgerMaxMarkDeleteRate;
+ "--ml-mark-delete-max-rate" },
+ description = "Throttling rate of mark-delete operation (0
means no throttle)")
+ private double managedLedgerMaxMarkDeleteRate = 0;
@Override
void run() throws PulsarAdminException {
String namespace = validateNamespace(params);
+ if (bookkeeperEnsemble <= 0 || bookkeeperWriteQuorum <= 0 ||
bookkeeperAckQuorum <= 0) {
+ throw new ParameterException("[--bookkeeper-ensemble],
[--bookkeeper-write-quorum] "
+ + "and [--bookkeeper-ack-quorum] must greater than
0.");
+ }
+ if (managedLedgerMaxMarkDeleteRate < 0) {
+ throw new ParameterException("[--ml-mark-delete-max-rate]
cannot less than 0.");
+ }
getAdmin().namespaces().setPersistence(namespace, new
PersistencePolicies(bookkeeperEnsemble,
bookkeeperWriteQuorum, bookkeeperAckQuorum,
managedLedgerMaxMarkDeleteRate));
}
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 c541cc1a575..255f2349710 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
@@ -1687,24 +1687,32 @@ public class CmdTopics extends CmdBase {
private java.util.List<String> params;
@Parameter(names = { "-e",
- "--bookkeeper-ensemble" }, description = "Number of bookies to
use for a topic", required = true)
- private int bookkeeperEnsemble;
+ "--bookkeeper-ensemble" }, description = "Number of bookies to
use for a topic")
+ private int bookkeeperEnsemble = 2;
@Parameter(names = { "-w",
- "--bookkeeper-write-quorum" }, description = "How many writes
to make of each entry", required = true)
- private int bookkeeperWriteQuorum;
+ "--bookkeeper-write-quorum" }, description = "How many writes
to make of each entry")
+ private int bookkeeperWriteQuorum = 2;
@Parameter(names = { "-a",
- "--bookkeeper-ack-quorum" }, description = "Number of acks
(guaranteed copies) to wait for each entry", required = true)
- private int bookkeeperAckQuorum;
+ "--bookkeeper-ack-quorum" }, description = "Number of acks
(guaranteed copies) to wait for each entry")
+ private int bookkeeperAckQuorum = 2;
@Parameter(names = { "-r",
- "--ml-mark-delete-max-rate" }, description = "Throttling rate
of mark-delete operation (0 means no throttle)", required = true)
- private double managedLedgerMaxMarkDeleteRate;
+ "--ml-mark-delete-max-rate" }, description = "Throttling rate
of mark-delete operation "
+ + "(0 means no throttle)")
+ private double managedLedgerMaxMarkDeleteRate = 0;
@Override
void run() throws PulsarAdminException {
String persistentTopic = validatePersistentTopic(params);
+ if (bookkeeperEnsemble <= 0 || bookkeeperWriteQuorum <= 0 ||
bookkeeperAckQuorum <= 0) {
+ throw new ParameterException("[--bookkeeper-ensemble],
[--bookkeeper-write-quorum] "
+ + "and [--bookkeeper-ack-quorum] must greater than
0.");
+ }
+ if (managedLedgerMaxMarkDeleteRate < 0) {
+ throw new ParameterException("[--ml-mark-delete-max-rate]
cannot less than 0.");
+ }
getTopics().setPersistence(persistentTopic, new
PersistencePolicies(bookkeeperEnsemble,
bookkeeperWriteQuorum, bookkeeperAckQuorum,
managedLedgerMaxMarkDeleteRate));
}
