zmiana okresu przechowywania kafka podczas wykonywania usługi

Question

W przypadku Kafki 0.8.1.1, jak zmienić czas przechowywania dziennika w trakcie jego działania? documentation mówi nieruchomość jest , ale próbuje to zmienić za pomocą kafka-topics.sh zwraca ten błąd

$ bin/kafka-topics.sh --zookeeper zk.yoursite.com --alter --topic as-access --config topic.log.retention.hours=24 
Error while executing topic command requirement failed: Unknown configuration "topic.log.retention.hours". 
java.lang.IllegalArgumentException: requirement failed: Unknown configuration "topic.log.retention.hours". 
    at scala.Predef$.require(Predef.scala:145) 
    at kafka.log.LogConfig$$anonfun$validateNames$1.apply(LogConfig.scala:138) 
    at kafka.log.LogConfig$$anonfun$validateNames$1.apply(LogConfig.scala:137) 
    at scala.collection.Iterator$class.foreach(Iterator.scala:631) 
    at scala.collection.JavaConversions$JEnumerationWrapper.foreach(JavaConversions.scala:479) 
    at kafka.log.LogConfig$.validateNames(LogConfig.scala:137) 
    at kafka.log.LogConfig$.validate(LogConfig.scala:145) 
    at kafka.admin.TopicCommand$.parseTopicConfigsToBeAdded(TopicCommand.scala:171) 
    at kafka.admin.TopicCommand$$anonfun$alterTopic$1.apply(TopicCommand.scala:95) 
    at kafka.admin.TopicCommand$$anonfun$alterTopic$1.apply(TopicCommand.scala:93) 
    at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:57) 
    at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:43) 
    at kafka.admin.TopicCommand$.alterTopic(TopicCommand.scala:93) 
    at kafka.admin.TopicCommand$.main(TopicCommand.scala:52) 
    at kafka.admin.TopicCommand.main(TopicCommand.scala) 

Answer 1

jest własnością maklera, który jest używany jako wartość domyślna podczas tworzenia tematu. Po zmianie konfiguracji aktualnie uruchomionego tematu za pomocą kafka-topics.sh należy określić właściwość poziomu tematu.

Właściwość na poziomie tematu dla czasu przechowywania dziennika to retention.ms.

Od Topic-level configuration w Kafka 0.8.1 dokumentacji:

• Property: retention.ms
• Domyślnie: 7 dni
• domyślny serwer Property: log.retention.minutes
• Opis: Ta konfiguracja kontroluje m Maksymalny czas, w którym zatrzymamy dziennik, zanim odrzucimy stare segmenty dziennika, aby zwolnić miejsce, jeśli używamy zasady przechowywania "usuń". Stanowi to umowę SLA o tym, jak szybko konsumenci będą musieli odczytać dane.

więc poprawna komenda jest

$ bin/kafka-topics.sh --zookeeper zk.yoursite.com --alter --topic as-access --config retention.ms=86400000 

Można sprawdzić, czy konfiguracja jest stosowana prawidłowo za pomocą następującego polecenia.

$ bin/kafka-topics.sh --describe --zookeeper zk.yoursite.com --topic as-access 

Następnie zobaczysz coś jak poniżej.

Topic:as-access PartitionCount:3 ReplicationFactor:3 Configs:retention.ms=86400000 

Answer 2

Prawidłowy klucz config jest retention.ms

$ bin/kafka-topics.sh --zookeeper zk.prod.yoursite.com --alter --topic as-access --config retention.ms=86400000 
Updated config for topic "my-topic". 

Answer 3

Poniżej jest właściwym sposobem, aby zmienić temat config jak z Kafki 0.10.2.0:

bin/kafka-configs.sh --zookeeper <zk_host> --alter --entity-type topics --entity-name test_topic --add-config retention.ms=86400000 

Temat config zmienić operacje zostały przestarzałe dla bin/kafka-topics.sh.

WARNING: Altering topic configuration from this script has been deprecated and may be removed in future releases. 
    Going forward, please use kafka-configs.sh for this functionality`