Skip to content

Commit

Permalink
[fix][cli] Fix set-retention with >2GB size value for topic policy (a…
Browse files Browse the repository at this point in the history
  • Loading branch information
lhotari authored and hanmz committed Feb 12, 2025
1 parent 4fd4f24 commit 7789ce2
Show file tree
Hide file tree
Showing 5 changed files with 65 additions and 11 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,6 @@
import java.util.function.Supplier;
import java.util.stream.Collectors;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.cli.converters.picocli.ByteUnitToIntegerConverter;
import org.apache.pulsar.cli.converters.picocli.ByteUnitToLongConverter;
import org.apache.pulsar.cli.converters.picocli.TimeUnitToMillisConverter;
import org.apache.pulsar.cli.converters.picocli.TimeUnitToSecondsConverter;
Expand Down Expand Up @@ -546,8 +545,8 @@ private class SetRetention extends CliCommand {
+ "For example, 4096, 10M, 16G, 3T. The size unit suffix character can be k/K, m/M, g/G, or t/T. "
+ "If the size unit suffix is not specified, the default unit is bytes. "
+ "0 or less than 1MB means no retention and -1 means infinite size retention", required = true,
converter = ByteUnitToIntegerConverter.class)
private Integer sizeLimit;
converter = ByteUnitToLongConverter.class)
private Long sizeLimit;

@Option(names = { "--global", "-g" }, description = "Whether to set this policy globally. "
+ "If set to true, the policy is replicated to other clusters asynchronously, "
Expand All @@ -560,8 +559,8 @@ void run() throws PulsarAdminException {
final int retentionTimeInMin = retentionTimeInSec != -1
? (int) TimeUnit.SECONDS.toMinutes(retentionTimeInSec)
: retentionTimeInSec.intValue();
final int retentionSizeInMB = sizeLimit != -1
? (int) (sizeLimit / (1024 * 1024))
final long retentionSizeInMB = sizeLimit != -1
? (sizeLimit / (1024 * 1024))
: sizeLimit;
getTopicPolicies(isGlobal).setRetention(persistentTopic,
new RetentionPolicies(retentionTimeInMin, retentionSizeInMB));
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1858,17 +1858,17 @@ private class SetRetention extends CliCommand {
+ "For example, 4096, 10M, 16G, 3T. The size unit suffix character can be k/K, m/M, g/G, or t/T. "
+ "If the size unit suffix is not specified, the default unit is bytes. "
+ "0 or less than 1MB means no retention and -1 means infinite size retention", required = true,
converter = ByteUnitToIntegerConverter.class)
private Integer sizeLimit;
converter = ByteUnitToLongConverter.class)
private Long sizeLimit;

@Override
void run() throws PulsarAdminException {
String persistentTopic = validatePersistentTopic(topicName);
final int retentionTimeInMin = retentionTimeInSec != -1
? (int) TimeUnit.SECONDS.toMinutes(retentionTimeInSec)
: retentionTimeInSec.intValue();
final int retentionSizeInMB = sizeLimit != -1
? (int) (sizeLimit / (1024 * 1024))
final long retentionSizeInMB = sizeLimit != -1
? (sizeLimit / (1024 * 1024))
: sizeLimit;
getTopics().setRetention(persistentTopic, new RetentionPolicies(retentionTimeInMin, retentionSizeInMB));
}
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,47 @@
/*
* 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 static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.TopicPolicies;
import org.apache.pulsar.common.policies.data.RetentionPolicies;
import org.testng.annotations.Test;

public class CmdTopicPoliciesTest {

@Test
public void testSetRetentionCmd() throws Exception {
TopicPolicies topicPolicies = mock(TopicPolicies.class);

PulsarAdmin admin = mock(PulsarAdmin.class);
when(admin.topicPolicies(anyBoolean())).thenReturn(topicPolicies);

CmdTopicPolicies cmd = new CmdTopicPolicies(() -> admin);

cmd.run("set-retention public/default/topic -s 2T -t 200d".split("\\s+"));

verify(topicPolicies, times(1)).setRetention("persistent://public/default/topic",
new RetentionPolicies(200 * 24 * 60, 2 * 1024 * 1024));
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,8 @@ public void testSetRetentionCmd() throws Exception {

CmdNamespaces cmd = new CmdNamespaces(() -> admin);

cmd.run("set-retention public/default -s 2T -t 2h".split("\\s+"));
verify(namespaces, times(1)).setRetention("public/default", new RetentionPolicies(120, 2 * 1024 * 1024));
cmd.run("set-retention public/default -s 2T -t 200d".split("\\s+"));
verify(namespaces, times(1)).setRetention("public/default",
new RetentionPolicies(200 * 24 * 60, 2 * 1024 * 1024));
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,7 @@
import org.apache.pulsar.client.impl.MessageIdImpl;
import org.apache.pulsar.common.naming.TopicDomain;
import org.apache.pulsar.common.policies.data.ManagedLedgerInternalStats.LedgerInfo;
import org.apache.pulsar.common.policies.data.RetentionPolicies;
import org.mockito.Mockito;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
Expand Down Expand Up @@ -260,4 +261,10 @@ public void testRunDeleteTopicsFromFileWithException() throws PulsarAdminExcepti
mockTopics = mock(Topics.class);
}

@Test
public void testSetRetentionCmd() throws Exception {
cmdTopics.run("set-retention public/default/topic -s 2T -t 200d".split("\\s+"));
verify(mockTopics, times(1)).setRetention("persistent://public/default/topic",
new RetentionPolicies(200 * 24 * 60, 2 * 1024 * 1024));
}
}

0 comments on commit 7789ce2

Please sign in to comment.