[
https://issues.apache.org/jira/browse/KAFKA-18818?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17936545#comment-17936545
]
José Armando García Sancio commented on KAFKA-18818:
----------------------------------------------------
[~ritvikgupta] thanks for the issue and your questions.
re 1:
The active controller is the leader of the kraft controller cluster and handle
all Admin RPCs. Admin RPCs like create topic write data to the
__cluster_metadata-0 partition that data is replicated to all of the inactive
controllers and brokers in the cluster using the FETCH RPC. The FETCH RPC is a
long polling RPC. If there is no data to replicate to the brokers or inactive
controllers, the active controller holds on to the request and doesn't send an
empty response until 500ms have expired.
The FETCH response contains both new data that needs to get replicated and the
HWM computed by the active controller that also needs to get replicated. The
issue is that the active controller only replies to FETCH request immediately,
if there is data to replica not if there is HWM to replicate (or update).
In practice this results in the follow flow of RPCs.
# Active controller handles create topic and writes data to
__cluster_metadata-0
# All of the replicas (inactive controller and broker) receive a FETCH
response with the new data. The HWM cannot be updated because the active
controller doesn't know that they have been successfully replicated to the
majority of the controllers.
# All of the brokers and inactive controllers send a new FETCH request. The
active controller updates the HWM but all of the FETCH RPC are held for 500ms
because there is no new data.
# If there is no new data, after 500ms the leader response to all of the held
FETCH RPC with no data but with the updated HWM.
re 3:
Yes. Based on my rough workflow above, my solution will remove the 500ms wait
time when there HWM that needs to get replicated. So you should expect HWM
propagation to be in the order of 3 x the network latency + 2 x disk write
latency. That's because the active controller needs the FETCH response with the
new data, and the brokers need to send a new FETCH request and finally the
active controller sends a FETCH response with the updated HWM.
Hope that helps.
> Significant delay in Metadata Update After Admin Client Operations in KRaft
> Mode
> --------------------------------------------------------------------------------
>
> Key: KAFKA-18818
> URL: https://issues.apache.org/jira/browse/KAFKA-18818
> Project: Kafka
> Issue Type: Bug
> Components: kraft
> Affects Versions: 3.9.0
> Reporter: Shubham Raj
> Priority: Major
>
> We have observed a significant delay in metadata updates after performing
> asynchronous Admin Client operations (such as {{{}createTopics{}}},
> {{{}deleteTopics{}}}, {{{}createAcls{}}}, and {{{}deleteAcls,
> changeTopicConfigs{}}}) using both the Confluent Python library and the
> Apache Kafka Java library when running Kafka in KRaft mode.
> *Issue Details:*
> When using the Admin Client to perform operations, we receive a {{Future}}
> map as the calls are asynchronous. Upon waiting for the {{Future}} results,
> we expect the operations to succeed, and the changes should be immediately
> reflected in the metadata. This behavior is consistent with what we observe
> when running Kafka in ZooKeeper mode.
> However, in KRaft mode, even after {{future.result()}} completes, the
> operation results are not instantly reflected in the metadata. We have
> observed an additional delay before the metadata is updated to reflect the
> changes. In our case, this delay is approximately 500ms, but it could vary
> depending on the cluster load.
> *Impact:*
> This delay is affecting our workflows where Kafka users need to perform Admin
> Client operations and expect immediate metadata updates. The additional wait
> time introduces latency and could potentially impact time-sensitive
> operations.
> *Steps performed :*
> * Called {{AdminClient.createTopics}} to create a test topic and waited on
> its KafkaFuture result, the future result returns null to denote a topic
> creation ( *[CreateTopicsResult
> ref|https://kafka.apache.org/23/javadoc/org/apache/kafka/clients/admin/CreateTopicsResult.html]*
> ).
> * Started a loop to try and describe the newly created topic ( to validate
> that the topic is present in Kafka's metadata ).
> Noted that the topic does not get immediately reflected in metadata, as we
> receive an _*{{UnknownTopicOrPartitionException}}*_
> * Checked and noted the number of iteration cycles it takes to properly
> reflect the newly created topic in metadata.
> In the below test the createTopics operation takes *407 ms* for the topic to
> properly appear in metadata ( which is additional time client needs to wait
> after the KafkaFuture resolves ).
>
> * *Kafka version: 3.9.0*
> * *Java library version: 2.8.0*
> {code:java}
> package kafkautils.trycode;
> import java.util.*;
> import local.kafka.KafkaConfig;
> import java.util.concurrent.Callable;
> import java.util.concurrent.ExecutionException;
> import java.util.function.Function;
> import java.util.logging.Logger;
> import org.apache.kafka.clients.admin.*;
> import org.apache.kafka.common.acl.*;
> import org.apache.kafka.common.config.ConfigResource;
> import org.apache.kafka.common.resource.PatternType;
> import org.apache.kafka.common.resource.ResourcePattern;
> import org.apache.kafka.common.resource.ResourcePatternFilter;
> import org.apache.kafka.common.resource.ResourceType;
> public class TryAdminClient {
> /** The Logger. **/
> private static final Logger LOG
> =LogFactory.getLogger(TryAdminClient.class);
> private static final String CLUSTER = "alpha";
> private static final String MAIN_RESOURCE_PREFIX =
> "testkafka.test_cluster_operations_delay.java_client.";
> private static final Admin adminClient;
> static {
> Properties adminProps = new Properties();
> adminProps.put(
> AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
> KafkaConfig.makeKafkaConfig(CLUSTER).getBootStrapServersWithPort()
> ); adminProps.put("security.protocol", "SASL_PLAINTEXT");
> adminProps.put("sasl.mechanism", "GSSAPI");
> adminProps.put("sasl.kerberos.service.name", "kafka");
> adminProps.put("sasl.kerberos.init.cmd", "/usr/local/bin/skinit
> --quiet");
> adminProps.put("sun.security.jgss.native", "true");
> adminProps.put("sun.security.jgss.lib", "/usr/libexec/libgsswrap.so");
> adminProps.put("javax.security.auth.useSubjectCredsOnly", "false");
> adminProps.put("java.security.auth.login.config",
>
> "/data/middleware/kafkautils/etc/config/kafka/kafka_client_jaas.conf");
> adminClient = Admin.create(adminProps);
> }
> public static void main(String[] args) throws ExecutionException,
> InterruptedException
> {
> TryAdminClient client = new TryAdminClient();
> String topic = MAIN_RESOURCE_PREFIX +
> "single_topic_operation_with_admin_client_test.a.1";
> var newTopic = new NewTopic(topic, 1,
> (short)3).configs(Map.of("cleanup.policy", "delete"));
> var createTopicsResult =
> adminClient.createTopics(List.of(newTopic)).all().get();
> LOG.info("CreateTopics finished with result : " + createTopicsResult);
> long queryStartTimestamp = System.currentTimeMillis();
> while (true) {
> try {
> var describeTopicsResult =
> adminClient.describeTopics(List.of(topic)).all().get().get(topic);
> LOG.info("DescribeTopics returned result : " +
> describeTopicsResult);
> if (describeTopicsResult != null)
> break;
> } catch (ExecutionException err) {
> LOG.warning("DescribeTopics operation failed with error : " +
> err);
> }
> }
> LOG.info(String.format(
> "Describe operation took = %d ms",
> System.currentTimeMillis() - queryStartTimestamp
> ));
> }
> } {code}
>
> *logs we got*
> {code:java}
> [20250214 07:11:43.514 EST (main) kafkautils.trycode.TryAdminClient#main
> INFO] CreateTopics finished with result : null
> [20250214 07:11:43.538 EST (main) kafkautils.trycode.TryAdminClient#main
> WARNING] DescribeTopics operation failed with error :
> java.util.concurrent.ExecutionException:
> org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server
> does not host this topic-partition... repeated 146 times
> [20250214 07:11:43.938 EST (main) kafkautils.trycode.TryAdminClient#main
> INFO] DescribeTopics returned result :
> (name=testkafka.test_cluster_operations_delay.java_client.single_topic_operation_with_admin_client_test.a.1,
> internal=false, partitions=(partition=0,
> leader=qtekafka-alpha-02.tbd.local.com:9092 (id: 6 rack: tbd.local.com),
> replicas=qtekafka-alpha-02.tbd.local.com:9092 (id: 6 rack: tbd.local.com),
> qtekafka-alpha-02.dr.local.com:9092 (id: 4 rack: dr.local.com),
> qtekafka-alpha-02.nyc.local.com:9092 (id: 2 rack: nyc.local.com),
> isr=qtekafka-alpha-02.tbd.local.com:9092 (id: 6 rack: tbd.local.com),
> qtekafka-alpha-02.dr.local.com:9092 (id: 4 rack: dr.local.com),
> qtekafka-alpha-02.nyc.local.com:9092 (id: 2 rack: nyc.local.com)),
> authorizedOperations=null)
> [20250214 07:11:43.939 EST (main) kafkautils.trycode.TryAdminClient#main
> INFO] Describe operation took = 407 ms{code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
