func (t *TopicApplier) Apply(ctx context.Context) error {
	log.Info("Validating configs...")
	brokerRacks := admin.DistinctRacks(t.brokers)
	if err := t.clusterConfig.Validate(); err != nil {
		return err
	}
	if err := t.topicConfig.Validate(len(brokerRacks)); err != nil {
		return err
	}
	if err := config.CheckConsistency(t.topicConfig, t.clusterConfig); err != nil {
		return err
	}
	log.Info("Checking if topic already exists...")
	topicInfo, err := t.adminClient.GetTopic(ctx, t.topicName, true)
	if err != nil {
		if err == admin.ErrTopicDoesNotExist {
			return t.applyNewTopic(ctx)
		}
		return err
	}
	return t.applyExistingTopic(ctx, topicInfo)
}
Apply runs a single "apply" run on the configured topic. The general flow is:

1. Validate configs
2. Acquire topic lock
3. Check if topic already exists
4. If new:
a. Create the topic
b. Update the placement in accordance with the configured strategy
5. If exists:
a. Check retention and update if needed
b. Check replication factor (can't be updated by topicctl)
c. Check partition count and extend if needed
d. Check partition placement and update/migrate if needed
e. Check partition leaders and update if needed
package apply
import (
	"context"
	"errors"
	"fmt"
	"path/filepath"
	"reflect"
	"strings"
	"time"
	"github.com/hashicorp/go-multierror"
	"github.com/segmentio/kafka-go"
	"github.com/segmentio/topicctl/pkg/admin"
	"github.com/segmentio/topicctl/pkg/apply/assigners"
	"github.com/segmentio/topicctl/pkg/apply/extenders"
	"github.com/segmentio/topicctl/pkg/apply/pickers"
	"github.com/segmentio/topicctl/pkg/apply/rebalancers"
	"github.com/segmentio/topicctl/pkg/config"
	"github.com/segmentio/topicctl/pkg/util"
	"github.com/segmentio/topicctl/pkg/zk"
	log "github.com/sirupsen/logrus"
)
type TopicApplierConfig struct {
	BrokerThrottleMBsOverride  int
	BrokersToRemove            []int
	ClusterConfig              config.ClusterConfig
	DryRun                     bool
	PartitionBatchSizeOverride int
	Rebalance                  bool
	RetentionDropStepDuration  time.Duration
	SkipConfirm                bool
	SleepLoopDuration          time.Duration
	TopicConfig                config.TopicConfig
}
type TopicApplier struct {
	config      TopicApplierConfig
	adminClient admin.Client
	brokers     []admin.BrokerInfo
	clusterConfig config.ClusterConfig
	maxBatchSize  int
	throttleBytes int64
	topicConfig   config.TopicConfig
	topicName     string
}
func NewTopicApplier(
	ctx context.Context,
	adminClient admin.Client,
	applierConfig TopicApplierConfig,
) (*TopicApplier, error) {
	if !adminClient.GetSupportedFeatures().Applies {
		return nil,
			errors.New(
				"Admin client does not support features needed for apply; please use zk-based client instead.",
			)
	}
	brokers, err := adminClient.GetBrokers(ctx, nil)
	if err != nil {
		return nil, err
	}
	var maxBatchSize int
	if applierConfig.PartitionBatchSizeOverride > 0 {
		maxBatchSize = applierConfig.PartitionBatchSizeOverride
	} else {
		maxBatchSize = applierConfig.TopicConfig.Spec.MigrationConfig.PartitionBatchSize
	}
	var throttleBytes int64
	if applierConfig.BrokerThrottleMBsOverride > 0 {
		throttleBytes = int64(applierConfig.BrokerThrottleMBsOverride) * 1000000
	} else if applierConfig.TopicConfig.Spec.MigrationConfig.ThrottleMB > 0 {
		throttleBytes = applierConfig.TopicConfig.Spec.MigrationConfig.ThrottleMB * 1000000
	} else if applierConfig.ClusterConfig.Spec.DefaultThrottleMB > 0 {
		throttleBytes = applierConfig.ClusterConfig.Spec.DefaultThrottleMB * 1000000
	} else {
		throttleBytes = 120000000
	}
	return &TopicApplier{
		adminClient:   adminClient,
		config:        applierConfig,
		brokers:       brokers,
		clusterConfig: applierConfig.ClusterConfig,
		maxBatchSize:  maxBatchSize,
		throttleBytes: throttleBytes,
		topicConfig:   applierConfig.TopicConfig,
		topicName:     applierConfig.TopicConfig.Meta.Name,
	}, nil
}