This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.

Status

Current state: ["Under DiscussionAccepted"]

Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]

JIRA: here [Change the link from KAFKA-1 to your own ticket]

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

Describe the problems you are trying to solve.

Public Interfaces

Briefly list any new interfaces that will be introduced as part of this proposal or any existing interfaces that will be removed or changed. The purpose of this section is to concisely call out the public contract that will come along with this feature.

A public interface is any change to the following:

• Binary log format

• The network protocol and api behavior

• Any class in the public packages under clientsConfiguration, especially client configuration

  • org/apache/kafka/common/serialization

  • org/apache/kafka/common

  • org/apache/kafka/common/errors

  • org/apache/kafka/clients/producer

  • org/apache/kafka/clients/consumer (eventually, once stable)

• Monitoring

• Command line tools and arguments

• Anything else that will likely break existing users in some way when they upgrade

Proposed Changes

...

KAFKA-17928 and KAFKA-17980

Motivation

RemoteLogManager uses multiple thread-pools to copy the segments, delete the expired segments, and read the data from remote storage. There can be situation to tune the number of threads in those pool to manage the incoming load. The proposal is to make them dynamic. 

Public Interfaces

• Make the below configs in the RemoteLogManager to be updated dynamically. Once they are updated, the respective thread-pool will be resized:
  • remote.log.manager.copier.thread.pool.size
  • remote.log.manager.expiration.thread.pool.size and
  • remote.log.reader.threads 
• Add a new method isReady in RemoteLogMetadataManager.

Proposed Changes

1. The thread-pools used by the RemoteLogManager does not allow to reconfigure the thread count dynamically. This is similar to updating the request handler thread counts dynamically and required to handle any unprecedented load in the remote storage. 
2. When a broker comes up online, the RemoteLogMetadataManager (RLMM) gets initialized in the background using the async fashion. The server starts to copy/delete the expired remote log segments, then the RLMM throws exception until it gets initialized. By adding the isReady method, the server can delay the copying (or) deleting the remote log segments until the RLMM gets initialized. This is a graceful handling to avoid exception/error logs while starting the server. 

[2024-11-06 09:33:16,867] WARN [RemoteLogManager=0 partition=xp3zuxhTQ9uUwk-uQhTx5g:topicA-0] Current task for topic-partition xp3zuxhTQ9uUwk-uQhTx5g:topicA-0 received error but it will be scheduled (kafka.log.remote.RemoteLogManager$RLMTask:793)
java.lang.IllegalStateException: This instance is in invalid state, initialized: false close: false
	at org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManager.ensureInitializedAndNotClosed(TopicBasedRemoteLogMetadataManager.java:557)
	at org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManager.highestOffsetForEpoch(TopicBasedRemoteLogMetadataManager.java:236)
	at kafka.log.remote.RemoteLogManager.findHighestRemoteOffset(RemoteLogManager.java:1827)
	at kafka.log.remote.RemoteLogManager$RLMFollowerTask.execute(RemoteLogManager.java:1434)
	at kafka.log.remote.RemoteLogManager$RLMTask.run(RemoteLogManager.java:784)

    /**
     * Denotes whether the partition metadata is ready to serve.
     *
     * @param topicIdPartition topic partition
     * @return True if the partition is ready to serve for remote storage operations.
     */
    default boolean isReady(TopicIdPartition topicIdPartition) {
        return true;
    }

Compatibility, Deprecation, and Migration Plan

• What impact (if any) will there be on existing users?
• If we are changing behavior how will we phase out the older behavior?
• If we need special migration tools, describe them here.
• When will we remove the existing behavior?

Test Plan

...

• The proposed changes are fully backward compatible. 

Test Plan

• Unit test to assert that the remote-log manager configs can be updated dynamically and thread-pool gets resized. 
• Unit test to assert that the copy/expiration logic won't be invoked for a partition until the RLMM gets ready to serve remote storage operations. 

Rejected Alternatives

If there are alternative ways of accomplishing the same thing, what were they? The purpose of this section is to motivate why the design is the way it is and not some other way.