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.
Authors: Ivan Yurchenko, Anatolii Popov
Current state: Under Discussion
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).
Kafka offers RemoteStorageManager interface as a way to let users plug a remote storage implementation. The broker also allows configuring the instance of the plugged RSM by forwarding to it the broker configuration prefixed with rsm.config. However, there’s neither a way for RemoteStorageManager to know the configuration of the topic with which segment it’s working, nor a way for the user to add custom topic configurations meaningful for a particular RSM. RemoteStorageManager implementations may benefit from this in situations where topic-specific tuning is needed e.g. for security (encryption enabled/disabled), data governance (the destination bucket or prefix), or performance.
This KIP proposes two changes that mostly make sense together.
Currently, when a topic configuration is changed (or a topic is created), the controller validates the configuration provided. If it detects unknown configuration keys, the whole operation is rejected. The KIP proposes the following:
unchecked.We propose to use the unchecked. prefix instead of something more specific to RSM in order for this mechanism to be more generic and potentially usable by other features and tools.
RemoteStorageManager methodsRemoteLogManager should pass the log configuration to RemoteStorageManager on each operation. This will allow a RSM implementation to benefit from the knowledge of the topic configurations, both known and custom (unchecked.) See Public Interfaces for the details of how the RemoteStorageManager needs to change.
There are two main changes to the public interfaces. The first change is to pass LogConfig to the methods of RemoteStorageManager with the default implementation that calls the existing methods (to preserve compatibility with the existing implementations).
default Optional<CustomMetadata> copyLogSegmentData(RemoteLogSegmentMetadata remoteLogSegmentMetadata,
LogSegmentData logSegmentData,
LogConfig logConfig)
throws RemoteStorageException {
return copyLogSegmentData(remoteLogSegmentMetadata, logSegmentData);
}
default InputStream fetchLogSegment(RemoteLogSegmentMetadata remoteLogSegmentMetadata,
int startPosition,
LogConfig logConfig) throws RemoteStorageException {
return fetchLogSegment(remoteLogSegmentMetadata, startPosition);
}
default InputStream fetchLogSegment(RemoteLogSegmentMetadata remoteLogSegmentMetadata,
int startPosition,
int endPosition
LogConfig logConfig) throws RemoteStorageException {
return fetchLogSegment(remoteLogSegmentMetadata, endPosition, startPosition);
}
default InputStream fetchIndex(RemoteLogSegmentMetadata remoteLogSegmentMetadata,
IndexType indexType,
LogConfig logConfig) throws RemoteStorageException {
return fetchIndex(remoteLogSegmentMetadata, indexType);
}
default void deleteLogSegmentData(RemoteLogSegmentMetadata remoteLogSegmentMetadata,
LogConfig logConfig) throws RemoteStorageException {
deleteLogSegmentData(remoteLogSegmentMetadata);
} |
The second public interface change is to update kafka-configs.sh and kafka-topics.sh so that they can handle the unchecked. configurations.
This feature does not introduce any incompatibility related to most of the Kafka ecosystem.
Existing RemoteStorageManager implementations will continue working due to the default implementations of the introduced methods.
The older versions of kafka-configs.sh and kafka-topics.sh are future-proof regarding this and just display unknown configurations as strings.
There may potentially exist third-party tools that break on an unknown topic configuration received from the API. However, these tools would be broken by adding any other topic configuration in a newer Kafka version, so the KIP doesn’t introduce a new incompatibility mode here.
The changes will be tested mostly on the unit level:
LogConfig, ControllerConfigurationValidatorTest, ConfigurationControlManagerTest, ReplicationControlManagerTest, TopicCommandTest, and ConfigCommandTest must be augmented to test unchecked configurations.RemoteLogManagerTest needs to be updated according to the interface changes and it needs to validate the topic configuration is passed.On the integration level:
BaseAdminIntegrationTest must be augmented to create and update topics with unchecked configs.There must be done some manual checks:
kafka-configs.sh and kafka-topics.sh must be able to correctly set and display unchecked configs.RemoteStorageManager compiled against the current stable version of the interface must be able to successfully run in a changed broker.There is a workaround that could be currently used. The RSM implementation that needs per-topic configuration could expect them to be made in its own configuration, for example, by a combination of lists, regular expressions, and so on. This doesn’t provide known (i.e. standard Kafka) topic configurations to the RSM, but at least allows custom settings to be made. This alternative is rejected for the following reasons: