1

Есть вот такой код:

fun main() {

    val props = Properties()
    props[StreamsConfig.APPLICATION_ID_CONFIG] = "first-stream-app"
    props[StreamsConfig.BOOTSTRAP_SERVERS_CONFIG] = "127.0.0.1:9092"
    props[StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG] = Serdes.String().javaClass
    props[StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG] = Serdes.String().javaClass

    val builder = StreamsBuilder()
    val textLines: KStream<String, String> = builder
                    .stream<String, String>("streams-demo", Consumed.with(Serdes.String(), Serdes.String()))

    fun getWords(line: String): List<String> =
            line.toLowerCase().split("\\W+".toRegex()).dropLastWhile { it.isEmpty() }
    val wordCounts: KTable<String, Long> = textLines
            .flatMapValues { textLine -> getWords(textLine) }
            .selectKey { _, value -> value }
            .groupByKey()
            .count(Materialized.`as`<String, Long, KeyValueStore<Bytes, ByteArray>>("counts-store"))
    wordCounts.toStream().to("streams-demo-out", Produced.with(Serdes.String(), Serdes.Long()))

    val streams = KafkaStreams(builder.build(), props)
    streams.start()

}

При запуске получаю кучу логов в таком формате:

21:04:57.939 [first-stream-app-1e30b3db-2237-4560-ac96-19af93aed5bb-StreamThread-1] DEBUG org.apache.kafka.clients.Metadata - [Consumer clientId=first-stream-app-1e30b3db-2237-4560-ac96-19af93aed5bb-StreamThread-1-consumer, groupId=first-stream-app] Updated cluster metadata updateVersion 32 to MetadataCache{cluster=Cluster(id = W4eDbi_qRD2udBrJu04BTA, nodes = [192.168.0.15:9092 (id: 0 rack: null)], partitions = [Partition(topic = streams-demo, partition = 0, leader = 0, replicas = [0], isr = [0], offlineReplicas = []), Partition(topic = first-stream-app-counts-store-repartition, partition = 0, leader = 0, replicas = [0], isr = [0], offlineReplicas = [])], controller = 192.168.0.15:9092 (id: 0 rack: null))}
21:04:58.039 [first-stream-app-1e30b3db-2237-4560-ac96-19af93aed5bb-StreamThread-1] DEBUG org.apache.kafka.clients.NetworkClient - [Consumer clientId=first-stream-app-1e30b3db-2237-4560-ac96-19af93aed5bb-StreamThread-1-consumer, groupId=first-stream-app] Sending metadata request MetadataRequestData(topics=null, allowAutoTopicCreation=true, includeClusterAuthorizedOperations=false, includeTopicAuthorizedOperations=false) to node 192.168.0.15:9092 (id: 0 rack: null)
21:04:58.042 [first-stream-app-1e30b3db-2237-4560-ac96-19af93aed5bb-StreamThread-1] DEBUG org.apache.kafka.clients.Metadata - [Consumer clientId=first-stream-app-1e30b3db-2237-4560-ac96-19af93aed5bb-StreamThread-1-consumer, groupId=first-stream-app] Updating last seen epoch from 0 to 0 for partition streams-demo-0
21:04:58.042 [first-stream-app-1e30b3db-2237-4560-ac96-19af93aed5bb-StreamThread-1] DEBUG org.apache.kafka.clients.Metadata - [Consumer clientId=first-stream-app-1e30b3db-2237-4560-ac96-19af93aed5bb-StreamThread-1-consumer, groupId=first-stream-app] Updating last seen epoch from 0 to 0 for partition first-stream-app-counts-store-repartition-0
21:04:58.042 [first-stream-app-1e30b3db-2237-4560-ac96-19af93aed5bb-StreamThread-1] DEBUG org.apache.kafka.clients.Metadata - [Consumer clientId=first-stream-app-1e30b3db-2237-4560-ac96-19af93aed5bb-StreamThread-1-consumer, groupId=first-stream-app] Updated cluster metadata updateVersion 33 to MetadataCache{cluster=Cluster(id = W4eDbi_qRD2udBrJu04BTA, nodes = [192.168.0.15:9092 (id: 0 rack: null)], partitions = [Partition(topic = streams-demo, partition = 0, leader = 0, replicas = [0], isr = [0], offlineReplicas = []), Partition(topic = first-stream-app-counts-store-repartition, partition = 0, leader = 0, replicas = [0], isr = [0], offlineReplicas = [])], controller = 192.168.0.15:9092 (id: 0 rack: null))}

Как отключить написание debug логов?

1 ответ 1

0

Просто настройте другой уровень логирования для org.apache.kafka. Поищите у себя в проекте файл log4j2.xml или log4j2.properties

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.