Function waitForReassignment

examples/alter_partition_reassignments/main.go:196–229

waitForReassignment polls ListPartitionReassignments until the broker stops reporting any in-progress reassignment for the topic, or the timeout fires. Kafka only returns partitions that are actively being reassigned, so an empty TopicStatus entry means the move is complete.

(
	ctx context.Context,
	admin sarama.ClusterAdmin,
	topic string,
	partitions []int32,
	interval, timeout time.Duration,
)

// Kafka only returns partitions that are actively being reassigned, so an
// empty TopicStatus entry means the move is complete.
func waitForReassignment(
	ctx context.Context,
	admin sarama.ClusterAdmin,
	topic string,
	partitions []int32,
	interval, timeout time.Duration,
) error {
	deadline := time.Now().Add(timeout)
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	for {
		status, err := admin.ListPartitionReassignments(topic, partitions)
		if err != nil {
			return fmt.Errorf("list partition reassignments: %w", err)
		}
		inProgress := status[topic]
		if len(inProgress) == 0 {
			return nil
		}
		log.Printf("%d partition(s) still reassigning", len(inProgress))
		for p, s := range inProgress {
			log.Printf("  partition %d: adding=%v removing=%v current=%v", p, s.AddingReplicas, s.RemovingReplicas, s.Replicas)
		}
		if time.Now().After(deadline) {
			return fmt.Errorf("timed out waiting for reassignment of %q to complete", topic)
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-ticker.C:
		}
	}
}

func partitionIDs(assignment [][]int32) []int32 {
	ids := make([]int32, len(assignment))