Merge branch 'segmentio:master' into shimon.support_fail_safe_approac…

…h_for_the_apply_cmd

.github/workflows/ci.yml

@@ -362,22 +362,9 @@ jobs:
           KAFKA_CFG_ADVERTISED_PORT: 9092
           KAFKA_CFG_AUTHORIZER_CLASS_NAME: kafka.security.authorizer.AclAuthorizer
           KAFKA_CFG_ALLOW_EVERYONE_IF_NO_ACL_FOUND: true
-
-  snyk:
-    runs-on: ubuntu-latest
-    needs: [test010, test270, test360]
-    steps:
-    - uses: actions/checkout@v3
-    - name: Run Snyk to check for vulnerabilities
-      uses: snyk/actions/golang@master
-      env:
-        SNYK_TOKEN: ${{ secrets.SNYK_TOKEN }}
-      with:
-        args: --severity-threshold=high --fail-on=upgradable
-
-
+
   publish-ghcr:
-    needs: [setup, snyk]
+    needs: [setup, test010, test270, test360]
     runs-on: ubuntu-latest
     if: ${{ ( github.ref_type == 'branch' ) && (( github.ref_name == 'master' ) || ( github.ref_name == 'v0' )) }}
     steps:
@@ -419,7 +406,7 @@ jobs:
     - run: echo "GHCR PUBLISH SUCCESSFUL"
 
   publish-dockerhub:
-    needs: [setup, snyk]
+    needs: [setup, test010, test270, test360]
     environment: CICD
     env: 
       RELEASE_TAG: ${{ needs.setup.outputs.version-tag }}

cmd/topicctl/subcmd/apply.go

@@ -35,6 +35,7 @@ type applyCmdConfig struct {
 	autoContinueRebalance        bool
 	retentionDropStepDurationStr string
 	skipConfirm                  bool
+	ignoreFewerPartitionsError   bool
 	sleepLoopDuration            time.Duration
 	failFast                     bool
 
@@ -100,6 +101,12 @@ func init() {
 		false,
 		"Skip confirmation prompts during apply process",
 	)
+	applyCmd.Flags().BoolVar(
+		&applyConfig.ignoreFewerPartitionsError,
+		"ignore-fewer-partitions-error",
+		false,
+		"Don't return error when topic's config specifies fewer partitions than it currently has",
+	)
 	applyCmd.Flags().DurationVar(
 		&applyConfig.sleepLoopDuration,
 		"sleep-loop-duration",
@@ -251,6 +258,7 @@ func applyTopic(
 			AutoContinueRebalance:      applyConfig.autoContinueRebalance,
 			RetentionDropStepDuration:  applyConfig.retentionDropStepDuration,
 			SkipConfirm:                applyConfig.skipConfirm,
+			IgnoreFewerPartitionsError: applyConfig.ignoreFewerPartitionsError,
 			SleepLoopDuration:          applyConfig.sleepLoopDuration,
 			TopicConfig:                topicConfig,
 		}

pkg/apply/apply.go

@@ -23,6 +23,8 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
+var ErrFewerPartitions = errors.New("fewer partitions in topic config")
+
 // TopicApplierConfig contains the configuration for a TopicApplier struct.
 type TopicApplierConfig struct {
 	BrokerThrottleMBsOverride  int
@@ -34,6 +36,7 @@ type TopicApplierConfig struct {
 	AutoContinueRebalance      bool
 	RetentionDropStepDuration  time.Duration
 	SkipConfirm                bool
+	IgnoreFewerPartitionsError bool
 	SleepLoopDuration          time.Duration
 	TopicConfig                config.TopicConfig
 }
@@ -213,6 +216,11 @@ func (t *TopicApplier) applyExistingTopic(
 	}
 
 	if err := t.updatePartitions(ctx, topicInfo); err != nil {
+		if errors.Is(err, ErrFewerPartitions) && t.config.IgnoreFewerPartitionsError {
+			log.Warnf("UpdatePartitions failure ignored. topic: %v, error: %v", t.topicName, err)
+			return nil
+		}
+
 		return err
 	}
 
@@ -477,7 +485,8 @@ func (t *TopicApplier) updatePartitions(
 
 	if currPartitions > t.topicConfig.Spec.Partitions {
 		return fmt.Errorf(
-			"Fewer partitions in topic config (%d) than observed (%d); this cannot be resolved by topicctl",
+			"%w (%d) than observed (%d); this cannot be resolved by topicctl",
+			ErrFewerPartitions,
 			t.topicConfig.Spec.Partitions,
 			currPartitions,
 		)