Function main

examples/alter_partition_reassignments/main.go:49–116

()

}

func main() {
	if verbose {
		sarama.Logger = log.New(os.Stdout, "[sarama] ", log.LstdFlags)
	}

	kafkaVersion, err := sarama.ParseKafkaVersion(version)
	if err != nil {
		log.Fatalf("parse kafka version: %v", err)
	}

	config := sarama.NewConfig()
	config.Version = kafkaVersion
	config.ClientID = "sarama-alter-partition-reassignments"

	admin, err := sarama.NewClusterAdmin(strings.Split(brokers, ","), config)
	if err != nil {
		log.Fatalf("create cluster admin: %v", err)
	}
	defer func() {
		if err := admin.Close(); err != nil {
			log.Printf("close cluster admin: %v", err)
		}
	}()

	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer cancel()

	current, err := describeTopic(admin, topic)
	if err != nil {
		log.Fatalf("describe topic %q: %v", topic, err)
	}
	log.Printf("current assignment for %q: %s", topic, formatAssignment(current))

	clusterBrokers, _, err := admin.DescribeCluster()
	if err != nil {
		log.Fatalf("describe cluster: %v", err)
	}
	if len(clusterBrokers) < replicationFactor {
		log.Fatalf("cluster has %d brokers, cannot satisfy replication factor %d", len(clusterBrokers), replicationFactor)
	}
	brokerIDs := make([]int32, 0, len(clusterBrokers))
	for _, b := range clusterBrokers {
		brokerIDs = append(brokerIDs, b.ID())
	}
	slices.Sort(brokerIDs)

	target, changed := buildTargetAssignment(current, brokerIDs, replicationFactor)
	if !changed {
		log.Printf("topic %q is already at replication factor %d, nothing to do", topic, replicationFactor)
		return
	}
	log.Printf("target assignment for %q:  %s", topic, formatAssignment(target))

	if err := admin.AlterPartitionReassignments(topic, target); err != nil {
		log.Fatalf("alter partition reassignments: %v", err)
	}
	log.Printf("submitted reassignment for %q, waiting for it to complete (timeout %s)", topic, pollTimeout)