kafka 스트림의 지연 기능

Question

kafka 스트림 코드로 뭔가를 시도하고 데이터를 분할 한 후 1ms 동안 지연 또는 thread.sleep()를 추가하려고했습니다 .... 어떻게해야합니까? 나 한테 그렇게 해?

KStreamBuilder builder = new KStreamBuilder(); 
KStream<String, String> textlines = builder.stream("INTOPIC"); 
KStream<String, String> mstream = textlines 
    .mapValues(value -> value.replace("[","")); 
    .mapValues(value -> value.replace("]","")); 
    .mapValues(value -> value.replaceAll("\\},\\{" ,"\\}\\},\\{\\{")) 
    .flatMapValues(value -> Arrays.asList(value.split("\\},\\{"))); 
mstream.to("OUTTOPIC"); 
KafkaStreams streams = new KafkaStreams(builder, config); 
streams.start(); 

그래서 .flatmapvalues ​​문 뒤에 나는 그래서이 내 문이 될 수 1ms의에 대한 Thread.sleep를()를 추가 할 필요가 ...?

Answer 1

달성하고자하는 것이 확실하지 않지만 처리 속도를 늦추고 싶으십니까? 사용 코드에 잠을 자면 충분합니다. 이를 위해 람다 식은 실제 결과를 반환하기 전에 "sleep"을 호출해야합니다. 또는 .foreach() 또는 peek() 전화를 추가하여 거기에서 잠을 자면됩니다.