Method IncrementalAlterConfig

admin.go:1008–1063

(resourceType ConfigResourceType, name string, entries map[string]IncrementalAlterConfigsEntry, validateOnly bool)

}

func (ca *clusterAdmin) IncrementalAlterConfig(resourceType ConfigResourceType, name string, entries map[string]IncrementalAlterConfigsEntry, validateOnly bool) error {
	var resources []*IncrementalAlterConfigsResource
	resources = append(resources, &IncrementalAlterConfigsResource{
		Type:          resourceType,
		Name:          name,
		ConfigEntries: entries,
	})

	request := &IncrementalAlterConfigsRequest{
		Resources:    resources,
		ValidateOnly: validateOnly,
	}

	if ca.conf.Version.IsAtLeast(V2_4_0_0) {
		request.Version = 1
	}

	var (
		b   *Broker
		err error
	)

	// AlterConfig of broker/broker logger must be sent to the broker in question
	if dependsOnSpecificNode(&ConfigResource{Name: name, Type: resourceType}) {
		var id int64
		id, err = strconv.ParseInt(name, 10, 32)
		if err != nil {
			return err
		}
		b, err = ca.findBroker(int32(id))
	} else {
		b, err = ca.findAnyBroker()
	}
	if err != nil {
		return err
	}

	_ = b.Open(ca.client.Config())
	rsp, err := b.IncrementalAlterConfigs(request)
	if err != nil {
		return err
	}

	for _, rspResource := range rsp.Resources {
		if rspResource.Name == name {
			if rspResource.ErrorCode != int16(ErrNoError) {
				err = KError(rspResource.ErrorCode)
				if rspResource.ErrorMsg != "" {
					err = fmt.Errorf("%w: %s", err, rspResource.ErrorMsg)
				}
				return err
			}
		}
	}
	return nil
}

func (ca *clusterAdmin) CreateACL(resource Resource, acl Acl) error {