feat(apply): add --keep-throttle flag to preserve replication throttles

Add a new `--keep-throttle` boolean flag to the apply command that allows
users to preserve replication throttle settings instead of having them
automatically removed after applying topic configurations.

This is particularly useful for topics where throttle settings
(leader.replication.throttled.replicas, follower.replication.throttled.replicas)
are intentionally configured in the topic YAML and should be maintained
across applies.

Changes:
- Add --keep-throttle flag to apply command
- Skip automatic throttle removal when flag is set
- Update help documentation

Example usage:
  topicctl apply --keep-throttle --skip-confirm \
    --cluster-config cluster.yaml topics/my-topic.yaml

Author: Robert Bernhof

README.md

@@ -124,7 +124,7 @@ docker-compose down
 #### apply
 
 ```
-topicctl apply [path(s) to topic config(s)]
+topicctl apply [flags] [path(s) to topic config(s)]
 ```
 
 The `apply` subcommand ensures that the actual state of a topic in the cluster
@@ -135,6 +135,28 @@ then the tool will initiate the necessary changes to bring it into compliance.
 See the [Config formats](#config-formats) section below for more information on the
 expected file formats.
 
+##### apply flags
+
+| Flag | Type | Default | Description |
+|------|------|---------|-------------|
+| `--skip-confirm` | bool | false | Skip confirmation prompts |
+| `--keep-throttle` | bool | false | Keep replication throttle settings instead of removing them after apply |
+| `--no-spinner` | bool | false | Disable progress spinner |
+| `--conn-timeout` | duration | 10s | Connection timeout |
+| `--rebalance` | bool | false | Rebalance topic replicas after apply |
+
+##### Apply Examples
+
+Apply while preserving replication throttle settings with skip confirm:
+```bash
+topicctl apply --skip-confirm --keep-throttle \
+  --cluster-config examples/local-cluster/cluster.yaml examples/local-cluster/topics/my-topic.yaml
+```
+
+This is useful for topics where throttle settings (e.g., `leader.replication.throttled.replicas`,
+`follower.replication.throttled.replicas`) are intentionally configured and should be maintained
+across applies.
+
 #### bootstrap
 
 ```

cmd/topicctl/subcmd/apply.go

@@ -39,6 +39,7 @@ type applyCmdConfig struct {
 	ignoreFewerPartitionsError   bool
 	sleepLoopDuration            time.Duration
 	failFast                     bool
+	keepThrottle                 bool
 
 	shared sharedOptions
 
@@ -120,6 +121,12 @@ func init() {
 		true,
 		"Fail upon the first error encountered during apply process",
 	)
+	applyCmd.Flags().BoolVar(
+		&applyConfig.keepThrottle,
+		"keep-throttle",
+		false,
+		"Keep replication throttle settings instead of removing them",
+	)
 
 	addSharedConfigOnlyFlags(applyCmd, &applyConfig.shared)
 	RootCmd.AddCommand(applyCmd)
@@ -263,6 +270,7 @@ func applyTopic(
 			IgnoreFewerPartitionsError: applyConfig.ignoreFewerPartitionsError,
 			SleepLoopDuration:          applyConfig.sleepLoopDuration,
 			TopicConfig:                topicConfig,
+			KeepThrottle:               applyConfig.keepThrottle,
 		}
 
 		if err := cliRunner.ApplyTopic(ctx, applierConfig); err != nil {

pkg/apply/apply.go

@@ -38,6 +38,7 @@ type TopicApplierConfig struct {
 	IgnoreFewerPartitionsError bool
 	SleepLoopDuration          time.Duration
 	TopicConfig                config.TopicConfig
+	KeepThrottle               bool
 }
 
 // TopicApplier executes an "apply" run on a topic by comparing the actual
@@ -281,7 +282,7 @@ func (t *TopicApplier) checkExistingState(
 			return nil
 		}
 
-		if topicInfo.IsThrottled() {
+		if topicInfo.IsThrottled() && !t.config.KeepThrottle {
 			log.Infof(
 				"It looks there are still throttles on the topic (config: %+v)",
 				topicInfo.Config,

pkg/apply/apply_test.go

@@ -901,6 +901,79 @@ func TestApplyOverrides(t *testing.T) {
 	assert.Equal(t, applier.maxBatchSize, 8)
 }
 
+func TestApplyKeepThrottle(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	topicName := util.RandomString("apply-keep-throttle-", 6)
+	topicConfig := config.TopicConfig{
+		Meta: config.ResourceMeta{
+			Name:        topicName,
+			Cluster:     "test-cluster",
+			Region:      "test-region",
+			Environment: "test-environment",
+		},
+		Spec: config.TopicSpec{
+			Partitions:        3,
+			ReplicationFactor: 2,
+			RetentionMinutes:  500,
+			PlacementConfig: config.TopicPlacementConfig{
+				Strategy: config.PlacementStrategyStatic,
+				Picker:   config.PickerMethodLowestIndex,
+				StaticAssignments: [][]int{
+					{1, 2},
+					{2, 3},
+					{1, 3},
+				},
+			},
+			MigrationConfig: &config.TopicMigrationConfig{
+				ThrottleMB:         2,
+				PartitionBatchSize: 3,
+			},
+		},
+	}
+	// Create topic
+	applier := testApplier(ctx, t, topicConfig)
+	defer applier.adminClient.Close()
+	err := applier.Apply(ctx)
+	require.NoError(t, err)
+
+	topicInfo, err := applier.adminClient.GetTopic(ctx, topicName, true)
+	require.NoError(t, err)
+	// Topic should not be throttled initially
+	assert.False(t, topicInfo.IsThrottled())
+
+	supported := applier.adminClient.GetSupportedFeatures()
+	if !(supported.Locks && supported.DynamicBrokerConfigs) {
+		// This test only works on zk-based clients for now
+		return
+	}
+
+	// Manually add throttles to the topic
+	_, err = applier.adminClient.UpdateTopicConfig(
+		ctx,
+		topicName,
+		[]kafka.ConfigEntry{
+			{
+				ConfigName:  admin.LeaderReplicasThrottledKey,
+				ConfigValue: "0:1,1:2,2:3",
+			},
+			{
+				ConfigName:  admin.FollowerReplicasThrottledKey,
+				ConfigValue: "0:2,1:3,2:1",
+			},
+		},
+		true,
+	)
+	require.NoError(t, err)
+
+	topicInfo, err = applier.adminClient.GetTopic(ctx, topicName, true)
+	require.NoError(t, err)
+	// Topic should now be throttled
+	assert.True(t, topicInfo.IsThrottled())
+
+	// Test 1: keepThrottle = false (default behavior - should remove throttles)
+}
+
 func testTopicName(name string) string {
 	return util.RandomString(fmt.Sprintf("topic-%s-", name), 6)
 }