KAFKA-15066: add "remote.log.metadata.manager.listener.name" config to rlmm #13828

showuon · 2023-06-08T10:34:30Z

add "remote.log.metadata.manager.listener.name" config to rlmm to allow producer/consumer to connect to the server. Also add tests.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

…o rlmm

divijvaidya · 2023-06-08T14:00:32Z

core/src/main/java/kafka/log/remote/RemoteLogManager.java

    private void configureRLMM() {
        final Map<String, Object> rlmmProps = new HashMap<>(rlmConfig.remoteLogMetadataManagerProps());

        rlmmProps.put(KafkaConfig.BrokerIdProp(), brokerId);
        rlmmProps.put(KafkaConfig.LogDirProp(), logDir);
+        endpoint.ifPresent(e -> {
+            rlmmProps.put("bootstrap.servers", e.host() + ":" + e.port());


please use the constant CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG

Yes, but that would add dependency with client module with core. That's why I use plain string instead

divijvaidya · 2023-06-08T14:01:23Z

core/src/main/java/kafka/log/remote/RemoteLogManager.java

    private void configureRLMM() {
        final Map<String, Object> rlmmProps = new HashMap<>(rlmConfig.remoteLogMetadataManagerProps());

        rlmmProps.put(KafkaConfig.BrokerIdProp(), brokerId);
        rlmmProps.put(KafkaConfig.LogDirProp(), logDir);
+        endpoint.ifPresent(e -> {
+            rlmmProps.put("bootstrap.servers", e.host() + ":" + e.port());
+            rlmmProps.put("security.protocol", e.securityProtocol().name);


please use the constant CommonClientConfigs.SECURITY_PROTOCOL_CONFIG

core/src/main/java/kafka/log/remote/RemoteLogManager.java

core/src/main/scala/kafka/server/KafkaServer.scala

core/src/main/scala/kafka/server/BrokerServer.scala

core/src/main/java/kafka/log/remote/RemoteLogManager.java

core/src/main/scala/kafka/server/KafkaServer.scala

core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java

divijvaidya

One last question Luke, should a unit test have failed when cluster.id property was missing in first revision? It probably hints towards a gap in testing. Can we add that? (or it's ok to say it will be added when Satish adds rest of the tests)

clients/src/main/java/org/apache/kafka/common/network/ListenerName.java

showuon · 2023-06-12T09:19:20Z

One last question Luke, should a unit test have failed when cluster.id property was missing in first revision? It probably hints towards a gap in testing. Can we add that? (or it's ok to say it will be added when Satish adds rest of the tests)

The test is added in the last commit: 39de62c

Also, I reverted the Endpoint back to Optional because I re-read the javadoc of RemoteLogMetadataManager:

kafka/storage/api/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogMetadataManager.java

Lines 45 to 48 in 513e1c6

    
            * <code>remote.log.metadata.manager.listener.name</code> property is about listener name of the local broker to which 
        
            * it should get connected if needed by RemoteLogMetadataManager implementation. When this is configured all other 
        
            * required properties can be passed as properties with prefix of 'remote.log.metadata.manager.listener. 
        
            * </p>

It's saying this is not a required config. Only used when needed by RemoteLogMetadataManager implementation. We should not throw exception if not provided.

divijvaidya

We probably want to update the KIP-405 here: https://cwiki.apache.org/confluence/display/KAFKA/KIP-405%3A+Kafka+Tiered+Storage#KIP405:KafkaTieredStorage-Configs and specify that this config is optional
From the javadoc of RLMM

When this is configured all other
 required properties can be passed as properties with prefix of 'remote.log.metadata.manager.listener

Can we please add a test to verify this? (asking because while constructing the rlmmProps, we don't pass any other configs with the listener prefix)

core/src/main/scala/kafka/server/BrokerServer.scala

satishd

Thanks @showuon for adding the missing integration of starting RLMM with the local endpoints of the broker. Had an initial pass of the PR, left a minor comment.

core/src/main/java/kafka/log/remote/RemoteLogManager.java

showuon · 2023-06-13T09:57:01Z

@divijvaidya @satishd , PR updated. Thanks.

We probably want to update the KIP-405 here: https://cwiki.apache.org/confluence/display/KAFKA/KIP-405%3A+Kafka+Tiered+Storage#KIP405:KafkaTieredStorage-Configs and specify that this config is optional

Updated.

From the javadoc of RLMM
When this is configured all other
required properties can be passed as properties with prefix of 'remote.log.metadata.manager.listener
Can we please add a test to verify this? (asking because while constructing the rlmmProps, we don't pass any other configs with the listener prefix)

I think we don't have this implemented. We should pass remote.log.metadata.* into RLMM based on KIP-405. Created KAFKA-15083 for this issue.

Thanks.

divijvaidya

Thanks for patiently working through all the comments Luke! Looks good to me.

satishd

Thanks @showuon for addressing the review comments. LGTM.

satishd · 2023-06-14T05:27:16Z

I think we don't have this implemented. We should pass remote.log.metadata.* into RLMM based on KIP-405. Created KAFKA-15083 for this issue.

@showuon This is no more valid, KIP needs to be updated with the prefix based configs for RSM and RLMM. Will update the KIP with those details.

showuon · 2023-06-14T12:13:46Z

I think we don't have this implemented. We should pass remote.log.metadata.* into RLMM based on KIP-405. Created KAFKA-15083 for this issue.

@showuon This is no more valid, KIP needs to be updated with the prefix based configs for RSM and RLMM. Will update the KIP with those details.

Good to know, thanks Satish. I've assigned KAFKA-15083 to you. You can close it once you've updated the KIP.

…AFKA-15066

showuon · 2023-06-16T03:46:11Z

Still cannot get a healthy CI build results. Try to rebase to the latest trunk.

showuon · 2023-06-16T12:48:57Z

Failed tests are unrelated and the failed testReplication also fail in trunk build. I've identified it's caused by this change: #13838 . Go ahead to merge it.

    Build / JDK 11 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest.testReplication()
    Build / JDK 8 and Scala 2.12 / org.apache.kafka.controller.QuorumControllerTest.testBalancePartitionLeaders()
    Build / JDK 17 and Scala 2.13 / kafka.admin.AddPartitionsTest.testIncrementPartitions(String).quorum=zk
    Build / JDK 17 and Scala 2.13 / org.apache.kafka.controller.QuorumControllerTest.testBalancePartitionLeaders()
    Build / JDK 17 and Scala 2.13 / org.apache.kafka.streams.integration.NamedTopologyIntegrationTest.shouldAllowRemovingAndAddingNamedTopologyToRunningApplicationWithMultipleNodesAndResetsOffsets()
    Build / JDK 11 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationTransactionsTest.testReplication()
    Build / JDK 11 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationTransactionsTest.testReplication()
    Build / JDK 8 and Scala 2.12 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationTransactionsTest.testReplication()
    Build / JDK 8 and Scala 2.12 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationTransactionsTest.testReplication()
    Build / JDK 17 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationTransactionsTest.testReplication()
    Build / JDK 17 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationTransactionsTest.testReplication()

jeqo · 2023-06-28T10:25:40Z

@showuon I'm trying to test this, but TBRLMM is still complaining about missing bootstrap.servers, even when listener name is provided:

kafka-ts  | [2023-06-28 10:19:04,131] INFO Initializing the resources. (org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManager)
kafka-ts  | [2023-06-28 10:19:04,141] ERROR Uncaught exception in thread 'RLMMInitializationThread': (org.apache.kafka.common.utils.KafkaThread)
kafka-ts  | org.apache.kafka.common.config.ConfigException: Missing required configuration "bootstrap.servers" which has no default value.
kafka-ts  |     at org.apache.kafka.common.config.ConfigDef.parseValue(ConfigDef.java:496)
kafka-ts  |     at org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:486)
kafka-ts  |     at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:112)
kafka-ts  |     at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:145)
kafka-ts  |     at org.apache.kafka.clients.admin.AdminClientConfig.<init>(AdminClientConfig.java:244)
kafka-ts  |     at org.apache.kafka.clients.admin.Admin.create(Admin.java:144)
kafka-ts  |     at org.apache.kafka.clients.admin.AdminClient.create(AdminClient.java:49)
kafka-ts  |     at org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManager.initializeResources(TopicBasedRemoteLogMetadataManager.java:366)
kafka-ts  |     at org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManager.lambda$configure$1(TopicBasedRemoteLogMetadataManager.java:352)
kafka-ts  |     at java.base/java.lang.Thread.run(Thread.java:829)

Looking at the code, I can see listener name being passed,

kafka-ts  |     remote.log.metadata.manager.class.name = org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManager
kafka-ts  |     remote.log.metadata.manager.class.path = null
kafka-ts  |     remote.log.metadata.manager.impl.prefix = rlmm.config.
kafka-ts  |     remote.log.metadata.manager.listener.name = BROKER

but when initializing the resources, properties without the right prefix are ignored:

kafka/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerConfig.java

Lines 136 to 159 in f32ebea

    
           private void initializeProducerConsumerProperties(Map<String, ?> configs) { 
        
               Map<String, Object> commonClientConfigs = new HashMap<>(); 
        
               Map<String, Object> producerOnlyConfigs = new HashMap<>(); 
        
               Map<String, Object> consumerOnlyConfigs = new HashMap<>(); 
        
               for (Map.Entry<String, ?> entry : configs.entrySet()) { 
        
                   String key = entry.getKey(); 
        
                   if (key.startsWith(REMOTE_LOG_METADATA_COMMON_CLIENT_PREFIX)) { 
        
                       commonClientConfigs.put(key.substring(REMOTE_LOG_METADATA_COMMON_CLIENT_PREFIX.length()), entry.getValue()); 
        
                   } else if (key.startsWith(REMOTE_LOG_METADATA_PRODUCER_PREFIX)) { 
        
                       producerOnlyConfigs.put(key.substring(REMOTE_LOG_METADATA_PRODUCER_PREFIX.length()), entry.getValue()); 
        
                   } else if (key.startsWith(REMOTE_LOG_METADATA_CONSUMER_PREFIX)) { 
        
                       consumerOnlyConfigs.put(key.substring(REMOTE_LOG_METADATA_CONSUMER_PREFIX.length()), entry.getValue()); 
        
                   } 
        
               } 
        
               HashMap<String, Object> allProducerConfigs = new HashMap<>(commonClientConfigs); 
        
               allProducerConfigs.putAll(producerOnlyConfigs); 
        
               producerProps = createProducerProps(allProducerConfigs); 
        
               HashMap<String, Object> allConsumerConfigs = new HashMap<>(commonClientConfigs); 
        
               allConsumerConfigs.putAll(consumerOnlyConfigs); 
        
               consumerProps = createConsumerProps(allConsumerConfigs); 
        
           }

Let me know if I'm reading this properly to create an issue, otherwise I may be missing something (is this supposed to be handled by https://issues.apache.org/jira/browse/KAFKA-15083). Many thanks!

cc @satishd @divijvaidya

showuon · 2023-06-30T08:32:57Z

@jeqo , thanks for raising it. Yes, I forgot to pass the bootstart.server as commonClientConfigs, so that it won't be passed into the producer/consumer. Are you available to file a small PR to fix it? I can work on that next week if you are busy. Thanks.

jeqo · 2023-06-30T10:17:48Z

Sure!, created this: #13938

KAFKA-15066: add "remote.log.metadata.manager.listener.name" config t…

f0cd237

…o rlmm

divijvaidya reviewed Jun 8, 2023

View reviewed changes

divijvaidya added the core label Jun 8, 2023

divijvaidya reviewed Jun 8, 2023

View reviewed changes

core/src/main/scala/kafka/server/BrokerServer.scala Outdated Show resolved Hide resolved

KAFKA-15066: add cluster.id into rlmm and refactor

3dd5775

divijvaidya reviewed Jun 9, 2023

View reviewed changes

core/src/main/java/kafka/log/remote/RemoteLogManager.java Show resolved Hide resolved

core/src/main/scala/kafka/server/KafkaServer.scala Outdated Show resolved Hide resolved

core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java Show resolved Hide resolved

KAFKA-15066: add tests for remote storage enabled case on server startup

25f90b1

divijvaidya reviewed Jun 9, 2023

View reviewed changes

clients/src/main/java/org/apache/kafka/common/network/ListenerName.java Outdated Show resolved Hide resolved

optional endpoint

39de62c

showuon force-pushed the KAFKA-15066 branch from 0dc097d to 39de62c Compare June 12, 2023 09:16

divijvaidya reviewed Jun 12, 2023

View reviewed changes

core/src/main/scala/kafka/server/BrokerServer.scala Outdated Show resolved Hide resolved

satishd reviewed Jun 12, 2023

View reviewed changes

core/src/main/java/kafka/log/remote/RemoteLogManager.java Outdated Show resolved Hide resolved

KAFKA-15066: address comments

4fc369d

divijvaidya approved these changes Jun 13, 2023

View reviewed changes

satishd approved these changes Jun 13, 2023

View reviewed changes

KAFKA-15066: add null check for remote iistener name

14470e6

showuon force-pushed the KAFKA-15066 branch from 6021158 to 14470e6 Compare June 15, 2023 01:47

divijvaidya added the tiered-storage Pull requests associated with KIP-405 (Tiered Storage) label Jun 15, 2023

showuon added 2 commits June 16, 2023 11:40

Merge branch 'trunk' of https://github.com/apache/kafka into KAFKA-15066

102faf9

Merge branch 'KAFKA-15066' of https://github.com/showuon/kafka into K…

4040fc1

…AFKA-15066

showuon merged commit 7423865 into apache:trunk Jun 16, 2023
1 check failed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-15066: add "remote.log.metadata.manager.listener.name" config to rlmm #13828

KAFKA-15066: add "remote.log.metadata.manager.listener.name" config to rlmm #13828

showuon commented Jun 8, 2023

divijvaidya Jun 8, 2023

showuon Jun 9, 2023

divijvaidya Jun 9, 2023

divijvaidya Jun 8, 2023

divijvaidya left a comment

showuon commented Jun 12, 2023

divijvaidya left a comment

satishd left a comment

showuon commented Jun 13, 2023 •

edited

divijvaidya left a comment

satishd left a comment

satishd commented Jun 14, 2023

showuon commented Jun 14, 2023

showuon commented Jun 16, 2023

showuon commented Jun 16, 2023

jeqo commented Jun 28, 2023 •

edited

showuon commented Jun 30, 2023

jeqo commented Jun 30, 2023

KAFKA-15066: add "remote.log.metadata.manager.listener.name" config to rlmm #13828

KAFKA-15066: add "remote.log.metadata.manager.listener.name" config to rlmm #13828

Conversation

showuon commented Jun 8, 2023

Committer Checklist (excluded from commit message)

divijvaidya Jun 8, 2023

Choose a reason for hiding this comment

showuon Jun 9, 2023

Choose a reason for hiding this comment

divijvaidya Jun 9, 2023

Choose a reason for hiding this comment

divijvaidya Jun 8, 2023

Choose a reason for hiding this comment

divijvaidya left a comment

Choose a reason for hiding this comment

showuon commented Jun 12, 2023

divijvaidya left a comment

Choose a reason for hiding this comment

satishd left a comment

Choose a reason for hiding this comment

showuon commented Jun 13, 2023 • edited

divijvaidya left a comment

Choose a reason for hiding this comment

satishd left a comment

Choose a reason for hiding this comment

satishd commented Jun 14, 2023

showuon commented Jun 14, 2023

showuon commented Jun 16, 2023

showuon commented Jun 16, 2023

jeqo commented Jun 28, 2023 • edited

showuon commented Jun 30, 2023

jeqo commented Jun 30, 2023

showuon commented Jun 13, 2023 •

edited

jeqo commented Jun 28, 2023 •

edited