[
https://issues.apache.org/jira/browse/KAFKA-18818?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18036341#comment-18036341
]
Ritvik Gupta commented on KAFKA-18818:
--------------------------------------
{quote}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.
{quote}
{quote}Before KAFKA-14145 or KAFKA-18751 is resolved (should be in v4.1.0), I
think setting the `metadata.max.idle.interval.ms` value to a lower value (ex:
50) should be able to workaround this problem. But of course, it'll increase
the metadata log size and the node resource usage.
{quote}
Thanks [~jsancio] & [~showuon] for the clarification and comments above ( just
got back to reading these out ). I see that this is to get fixed out now *v4.1*
( or *v4.2* ), so we will try to jump up our clusters to that version and test
out this fix for us.
> 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
> Assignee: José Armando García Sancio
> Priority: Major
> Fix For: 4.1.0, 4.2.0
>
> Attachments: old-batch-error.patch
>
>
> 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)
