Operator does not restart kafka controllers after tiered storage is enabled, preventing data offload into S3 bucket

[totochichi]

Hello all, we are using custom image with tiered storage plugin with the below config (Operator version 0.45 and Kafka 3.8.0)
We also have MirrorMaker2 installed in this namespace.
After helmchart deployment, kafka brokers restart and we are able to create topic with remote storage enabled, but the topics data are not being sent to S3 bucket.

 kafka:
      authorization:
        superUsers:
        ...
        type: simple
      config:
        auto.create.topics.enable: false
        default.replication.factor: 3
        log.retention.hours: 24
        min.insync.replicas: 2
        num.io.threads: 8
        num.network.threads: 3
        num.recovery.threads.per.data.dir: 1
        num.replica.fetchers: 1
        offsets.topic.replication.factor: 3
        principal.builder.class: cz.scholz.customprincipalbuilder.CustomPrincipalBuilder
        remote.log.manager.task.interval.ms: 30000
        remote.log.manager.thread.pool.size: 4
        remote.log.reader.threads: 5
        remote.log.storage.system.enable: true
        rsm.config.chunk.size: 4194304
        rsm.config.compression.enabled: true
        rsm.config.compression.heuristic.enabled: true
        rsm.config.storage.backend.class: io.aiven.kafka.tieredstorage.storage.s3.S3Storage
        rsm.config.storage.s3.bucket.name: bucket-kafka-example
        rsm.config.storage.s3.region: us-east-1
        rsm.config.upload.rate.limit.bytes.per.second: 104857600
        transaction.state.log.min.isr: 2
        transaction.state.log.replication.factor: 3
      image: strimzi-image/kafka:0.45.0-kafka-3.8.0-77
     tieredStorage:
        remoteStorageManager:
          className: io.aiven.kafka.tieredstorage.RemoteStorageManager
          classPath: /opt/kafka/plugins/tiered-storage-for-apache-kafka/core/*:/opt/kafka/plugins/tiered-storage-for-apache-kafka/s3/*
          config:
            aws.credentials.provider.class: software.amazon.awssdk.auth.credentials.WebIdentityTokenFileCredentialsProvider
            remote.log.metadata.manager.impl.prefix: rlmm.config.
            remote.log.storage.manager.impl.prefix: rsm.config.
        type: custom
      version: 3.8.0

Upon checking controller logs, we see the error below. If we delete the controllers and let them restart, then the errors disappear and data sent to S3 successfully. Do we know why this happens? Should the Operator restart Controllers as well if the cert or ACLs needs to be refreshed?

ERROR [ControllerApis nodeId=3] Unexpected error handling request
Forwarded request:
  RequestContext(
    header=RequestHeader(
      apiKey=ENVELOPE,
      apiVersion=0,
      clientId=2,
      correlationId=15,
      headerVersion=2
    ),
    connectionId='IP_A:PORT-TO-IP_B:PORT',
    clientAddress=/IP_B,
    principal=User:CN=kafka-kafka,O=ORG_PLACEHOLDER,
    listenerName=ListenerName(CONTROLPLANE-9090),
    securityProtocol=SSL,
    clientInformation=ClientInformation(
      softwareName=apache-kafka-java,
      softwareVersion=3.8.0
    ),
    fromPrivilegedListener=false,
    principalSerde=Optional[cz.scholz.customprincipalbuilder.CustomPrincipalBuilder@MASKED]
  )
  RequestHeader(
    apiKey=CREATE_ACLS,
    apiVersion=3,
    clientId=sourceCluster->targetCluster|sourceCluster->targetCluster.MirrorSourceConnector|replication-target-admin,
    correlationId=104,
    headerVersion=2
  )
  – CreateAclsRequestData(
      creations=[
        AclCreation(resourceType=2, resourceName='sourceCluster.heartbeats', resourcePatternType=3, principal='User:MASKED', host='*', operation=5, permissionType=3),
        ...
      ]
    )
  with context RequestContext(
    header=RequestHeader(
      apiKey=CREATE_ACLS,
      apiVersion=3,
      clientId=sourceCluster->targetCluster|sourceCluster->targetCluster.MirrorSourceConnector|replication-target-admin,
      correlationId=104,
      headerVersion=2
    ),
    connectionId='IP_A:PORT-TO-IP_C:PORT',
    clientAddress=/IP_C,
    principal=User:MASKED,
    listenerName=ListenerName(CONTROLPLANE-9090),
    securityProtocol=SSL,
    clientInformation=ClientInformation(
      softwareName=unknown,
      softwareVersion=unknown
    ),
    fromPrivilegedListener=false,
    principalSerde=Optional.empty
  )
  (kafka.server.ControllerApis) [data-plane-kafka-request-handler-4]

org.apache.kafka.common.errors.ClusterAuthorizationException:
  Request Request(
    processor=0,
    connectionId=IP_A:PORT-TO-IP_B:PORT,
    session=org.apache.kafka.network.Session@MASKED,
    listenerName=ListenerName(CONTROLPLANE-9090),
    securityProtocol=SSL,
    buffer=java.nio.HeapByteBuffer[pos=MASKED lim=MASKED cap=MASKED],
    envelope=Some(
      Request(
        processor=0,
        connectionId=IP_A:PORT-TO-IP_B:PORT,
        session=org.apache.kafka.network.Session@MASKED,
        listenerName=ListenerName(CONTROLPLANE-9090),
        securityProtocol=SSL,
        buffer=java.nio.HeapByteBuffer[pos=MASKED lim=MASKED cap=MASKED],
        envelope=None
      )
    )

[scholzj]

Your tiered storage configuration definitely does not look exactly as I would expect. It should look something like this: https://github.com/scholzj/demo-autoscaling-apache-kafka-with-strimzi/blob/main/01-kafka.yaml#L38-L47 -> the tiered storage configuration should be folded in the tiered storage section (different Aiven plugin, but the idea is the same). Having it outside might lead to things being ignored etc.

That said ...

• The controller-only nodes have nothing to do with the tiered storage, so there should be no need to roll them. But not 100% sure if that would be the case without the full logs.
• The controller errors do not seem to have anything to do with tiered storage. They seem to be related to a misconfigured Mirror Maker which is missing some privileges. (Also - while it is a bit hard to say without the full logs - I wonder if this is a limitation of the CustomPrincipalBuilder and how it works in KRaft 🤔)

[scholzj]

Hmm, have you tried to reproduce it with Kafka 4.0.0 as well? Where tiered storage is GA?

These messages seem strange:

2025-06-25 17:14:08,517 WARN [QuorumController id=100] Broker 1 registered with feature metadata.version that is unknown to the controller (org.apache.kafka.controller.ClusterControlManager) [quorum-controller-100-event-handler]
2025-06-25 17:14:51,925 WARN [QuorumController id=100] Broker 2 registered with feature metadata.version that is unknown to the controller (org.apache.kafka.controller.ClusterControlManager) [quorum-controller-100-event-handler]
2025-06-25 17:15:35,118 WARN [QuorumController id=100] Broker 0 registered with feature metadata.version that is unknown to the controller (org.apache.kafka.controller.ClusterControlManager) [quorum-controller-100-event-handler]

There is no reason for brokers with unknown metadata version. This one:

2025-06-25 20:37:01,436 INFO [QuorumController id=100] CreateTopics result(s): CreatableTopic(name='tieredstorage_test_topic', numPartitions=3, replicationFactor=3, assignments=[], configs=[CreateableTopicConfig(name='local.retention.ms', value='10000'), CreateableTopicConfig(name='remote.storage.enable', value='true'), CreateableTopicConfig(name='segment.bytes', value='1000000')]): INVALID_CONFIG (Tiered Storage functionality is disabled in the broker. Topic cannot be configured with remote log storage.) (org.apache.kafka.controller.ReplicationControlManager) [quorum-controller-100-event-handler]

Would suggest that the controller needs to be aware of the tiered storage to make it work.

How do you create the topics?

---

I think ultimately, someone, needs to try if the same problem can be reproduced with Strimzi 0.46 / Kafka 4.0. As a lot might have changed in the tiered storage support since Kafka 3.8 where it was not complete yet, if I remember correctly.

[totochichi]

This is the command I use to create topics. I tested this with Strimzi 0.46 and Kafka 4.0 today and it behaves exactly the same.

 confluent-7.8.0/bin/kafka-topics --command-config client.properties --create --partitions 3 --replication-factor 3 --config remote.storage.enable=true --config local.retention.ms=10000 --config segment.bytes=1000000 --bootstrap-server [hostname]:443  --topic tieredstorage_test_topic

[scholzj]

What is in the client.properties?

[scholzj]

Never mind, I do not need the client.properties.

---

I managed to reproduce this with Kafka 4.0 / Strimzi 0.46.1. I opened #11595 to have it tracked and fix it.

Thanks for reporting this! (And sorry for closing your original issue - it turns out it was justified 🤦)

[totochichi]

Apologies I couldn't follow up for the past few weeks. Appreciate the quick turnaround, thank you!