Method AlterPartitionReassignments

admin.go:681–727

(topic string, assignment [][]int32)

}

func (ca *clusterAdmin) AlterPartitionReassignments(topic string, assignment [][]int32) error {
	if topic == "" {
		return ErrInvalidTopic
	}

	request := &AlterPartitionReassignmentsRequest{
		TimeoutMs: int32(60000),
		Version:   int16(0),
	}

	for i := range assignment {
		request.AddBlock(topic, int32(i), assignment[i])
	}

	return ca.retryOnError(isRetriableControllerError, func() error {
		b, err := ca.Controller()
		if err != nil {
			return err
		}

		errs := make([]error, 0)

		rsp, err := b.AlterPartitionReassignments(request)

		if err != nil {
			errs = append(errs, err)
		} else {
			if rsp.ErrorCode > 0 {
				errs = append(errs, rsp.ErrorCode)
			}

			for topic, topicErrors := range rsp.Errors {
				for partition, partitionError := range topicErrors {
					if !errors.Is(partitionError.errorCode, ErrNoError) {
						errs = append(errs, fmt.Errorf("[%s-%d]: %w", topic, partition, partitionError.errorCode))
					}
				}
			}
		}

		if len(errs) > 0 {
			return Wrap(ErrReassignPartitions, errs...)
		}

		return nil
	})
}

func (ca *clusterAdmin) ListPartitionReassignments(topic string, partitions []int32) (topicStatus map[string]map[int32]*PartitionReplicaReassignmentsStatus, err error) {
	if topic == "" {