Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[improve][cli] Add --cleanupSubscription to pulsar-admin #20028

Merged
merged 3 commits into from
Apr 23, 2023
Merged

[improve][cli] Add --cleanupSubscription to pulsar-admin #20028

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3f1e386
[improve][cli] Add `--cleanupSubscription` to pulsar-admin
jiangpengcheng Apr 6, 2023
762dda1
Fix style
jiangpengcheng Apr 11, 2023
e35f427
Merge branch 'master' into add_cleanup_subscription
tisonkun Apr 23, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -201,6 +201,9 @@ abstract class FunctionDetailsCommand extends BaseCommand {
protected String className;
@Parameter(names = { "-t", "--function-type" }, description = "The built-in Pulsar Function type")
protected String functionType;
@Parameter(names = "--cleanup-subscription", description = "Whether delete the subscription "
+ "when function is deleted")
protected Boolean cleanupSubscription;
@Parameter(names = "--jar", description = "Path to the JAR file for the function "
+ "(if the function is written in Java). It also supports URL path [http/https/file "
+ "(file protocol assumes that file already exists on worker host)/function "
Expand Down Expand Up @@ -471,6 +474,10 @@ void processArguments() throws Exception {
}
}

if (null != cleanupSubscription) {
functionConfig.setCleanupSubscription(cleanupSubscription);
}

if (null != inputs) {
List<String> inputTopics = Arrays.asList(inputs.split(","));
functionConfig.setInputs(inputTopics);
Expand Down
8 changes: 8 additions & 0 deletions pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -291,6 +291,10 @@ abstract class SinkDetailsCommand extends BaseCommand {
@Parameter(names = { "-t", "--sink-type" }, description = "The sinks's connector provider")
protected String sinkType;

@Parameter(names = "--cleanup-subscription", description = "Whether delete the subscription "
+ "when sink is deleted")
protected Boolean cleanupSubscription;

@Parameter(names = { "-i",
"--inputs" }, description = "The sink's input topic or topics "
+ "(multiple topics can be specified as a comma-separated list)")
Expand Down Expand Up @@ -469,6 +473,10 @@ void processArguments() throws Exception {
sinkConfig.setProcessingGuarantees(processingGuarantees);
}

if (null != cleanupSubscription) {
sinkConfig.setCleanupSubscription(cleanupSubscription);
}

if (retainOrdering != null) {
sinkConfig.setRetainOrdering(retainOrdering);
}
Expand Down
10 changes: 0 additions & 10 deletions pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -202,12 +202,6 @@ public static FunctionDetails convert(SinkConfig sinkConfig, ExtractedSinkDetail
sourceSpecBuilder.setNegativeAckRedeliveryDelayMs(sinkConfig.getNegativeAckRedeliveryDelayMs());
}

if (sinkConfig.getCleanupSubscription() != null) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Removals on this file seems breaking change?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's duplicated in lines 196~200

other deletions in this file also because of duplicate code

sourceSpecBuilder.setCleanupSubscription(sinkConfig.getCleanupSubscription());
} else {
sourceSpecBuilder.setCleanupSubscription(true);
}

if (sinkConfig.getSourceSubscriptionPosition() == SubscriptionInitialPosition.Earliest) {
sourceSpecBuilder.setSubscriptionPosition(Function.SubscriptionPosition.EARLIEST);
} else {
Expand Down Expand Up @@ -329,7 +323,6 @@ public static SinkConfig convertFromDetails(FunctionDetails functionDetails) {
// Set subscription position
sinkConfig.setSourceSubscriptionPosition(
convertFromFunctionDetailsSubscriptionPosition(functionDetails.getSource().getSubscriptionPosition()));
sinkConfig.setCleanupSubscription(functionDetails.getSource().getCleanupSubscription());

if (functionDetails.getSource().getTimeoutMs() != 0) {
sinkConfig.setTimeoutMs(functionDetails.getSource().getTimeoutMs());
Expand Down Expand Up @@ -671,9 +664,6 @@ public static SinkConfig validateUpdate(SinkConfig existingConfig, SinkConfig ne
if (!StringUtils.isEmpty(newConfig.getCustomRuntimeOptions())) {
mergedConfig.setCustomRuntimeOptions(newConfig.getCustomRuntimeOptions());
}
if (newConfig.getCleanupSubscription() != null) {
mergedConfig.setCleanupSubscription(newConfig.getCleanupSubscription());
}
if (newConfig.getTransformFunction() != null) {
mergedConfig.setTransformFunction(newConfig.getTransformFunction());
}
Expand Down