Mise à jour de TTL pour un sujet particulier dans kafka en utilisant Java

Question

Mettez à jour TTL pour un sujet afin que les enregistrements restent dans le sujet pendant 10 jours. Je dois faire cela pour un sujet particulier en laissant tous les autres sujets TTL est la même configuration actuelle, je dois le faire en utilisant java parce que je pousse un sujet à kafka via Java. Je suis en train de propriétés suivantes pour pousser un sujet à kafka

Properties props = new Properties(); 
    props.put("bootstrap.servers", KAFKA_SERVERS); 
    props.put("acks", ACKS); 
    props.put("retries", RETRIES); 
    props.put("linger.ms", new Integer(LINGER_MS)); 
    props.put("buffer.memory", new Integer(BUFFER_MEMORY)); 
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 

Answer 1

Vous pouvez le faire en utilisant le AdminClient, après un bout de code qui obtiennent la configuration actuelle (juste pour le test) puis mettre à jour les « retention.ms » config sur le sujet nommé "test".

Properties props = new Properties(); 
    props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); 

    AdminClient adminClient = AdminClient.create(props); 

    ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, "test"); 

    // get the current topic configuration 
    DescribeConfigsResult describeConfigsResult = 
      adminClient.describeConfigs(Collections.singleton(resource)); 

    Map<ConfigResource, Config> config = describeConfigsResult.all().get(); 

    System.out.println(config); 

    // create a new entry for updating the retention.ms value on the same topic 
    ConfigEntry retentionEntry = new ConfigEntry(TopicConfig.RETENTION_MS_CONFIG, "50000"); 
    Map<ConfigResource, Config> updateConfig = new HashMap<ConfigResource, Config>(); 
    updateConfig.put(resource, new Config(Collections.singleton(retentionEntry))); 

    AlterConfigsResult alterConfigsResult = adminClient.alterConfigs(updateConfig); 
    alterConfigsResult.all(); 

    describeConfigsResult = adminClient.describeConfigs(Collections.singleton(resource)); 

    config = describeConfigsResult.all().get(); 

    System.out.println(config); 

    adminClient.close();