apply

Documentation

Index

• Variables
• func FormatMissingKeys(configMap map[string]string, missingKeys []string) string
• func FormatNewTopicConfig(config kafka.TopicConfig) string
• func FormatSettingsDiff(topicSettings config.TopicSettings, configMap map[string]string, ...) (string, error)
• type TopicApplier
  • func NewTopicApplier(ctx context.Context, adminClient admin.Client, ...) (*TopicApplier, error)
  • func (t *TopicApplier) Apply(ctx context.Context) error
• type TopicApplierConfig

Variables

var ErrFewerPartitions = errors.New("fewer partitions in topic config")

Functions

func FormatMissingKeys

func FormatMissingKeys(
	configMap map[string]string,
	missingKeys []string,
) string

FormatMissingKeys generates a table that summarizes the key/value pairs that are set in the config in ZK but missing from the topic config.

func FormatNewTopicConfig

func FormatNewTopicConfig(config kafka.TopicConfig) string

FormatNewTopicConfig generates a pretty string representation of a kafka-go topic config.

func FormatSettingsDiff

func FormatSettingsDiff(
	topicSettings config.TopicSettings,
	configMap map[string]string,
	diffKeys []string,
) (string, error)

FormatSettingsDiff generates a table that summarizes the differences between the topic settings from a topic config and the settings from ZK.

Types

type TopicApplier

type TopicApplier struct {
	// contains filtered or unexported fields
}

TopicApplier executes an "apply" run on a topic by comparing the actual and desired configurations, and then updating the topic as necessary to align the two.

func NewTopicApplier

func NewTopicApplier(
	ctx context.Context,
	adminClient admin.Client,
	applierConfig TopicApplierConfig,
) (*TopicApplier, error)

NewTopicApplier creates and returns a new TopicApplier instance.

func (*TopicApplier) Apply

func (t *TopicApplier) Apply(ctx context.Context) error

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

type TopicApplierConfig

type TopicApplierConfig struct {
	BrokerThrottleMBsOverride  int
	BrokersToRemove            []int
	ClusterConfig              config.ClusterConfig
	DryRun                     bool
	PartitionBatchSizeOverride int
	Rebalance                  bool
	AutoContinueRebalance      bool
	RetentionDropStepDuration  time.Duration
	SkipConfirm                bool
	IgnoreFewerPartitionsError bool
	SleepLoopDuration          time.Duration
	TopicConfig                config.TopicConfig
}

TopicApplierConfig contains the configuration for a TopicApplier struct.