Блог

Я создал конечную точку отдыха, откуда поступают данные, и я отправляю их в тему кафки «тема-обратная связь» . Теперь я хочу обработать эти данные и отправить их в другую тему кафки «тема-хорошее слово». Теперь из конечной точки rest данные поступают в раздел «тема-обратная связь» правильно , но после их обработки и отправки в другую тему «тема-хорошее слово» данные не отображаются в разделе «тема-хорошее слово».

В программе нет ошибок

Вот мой код:

 @Configuration
public class FeedbackStream {
    
    Set<String> GOOD_WORDS = Set.of("happy", "good", "helpful");

    @Bean
    public KStream<String,String> KStreamFeedback(StreamsBuilder builder)
    {
        var stringSerde = Serdes.String();
        var feedbackSerde = new JsonSerde<>(Feedback.class);

        ((JsonDeserializer) feedbackSerde.deserializer()).setUseTypeHeaders(false);
        
         var sourceStream = builder.stream("topic-feedback", Consumed.with(stringSerde,feedbackSerde))
                                                        .flatMapValues(mapperGoodWords());
          
        sourceStream.to("topic-goodWord", Produced.with(stringSerde,stringSerde));

        return sourceStream;
    }





private ValueMapper<Feedback,Iterable<String> > mapperGoodWords() 
    {
        return feedback -> Arrays
                           .asList(feedback.getFeedback().replaceAll("[^a-zA-Z]", "").toLowerCase()
                           .split("\s ")).stream()
                           .filter(word -> GOOD_WORDS.contains(word)).distinct().collect(Collectors.toList());
    }

    
 

Feedback.java

 public class Feedback {
    
    private String location;
    private int rating;
    private String feedback;     
    
    // getters,setters
}
  
 
 

конфигурация потока кафки

 @Configuration
@EnableKafkaStreams
public class KafkaStreamConfig {
    
    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
    public KafkaStreamsConfiguration kafkaStreamConfig()
    {
        var props = new HashMap<String, Object>();

        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "kafka-stream");
      
         props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");

        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,Serdes.String().getClass().getName());
        props.put(JsonDeserializer.USE_TYPE_INFO_HEADERS,"false");
      
        return new KafkaStreamsConfiguration(props);
    }
}
 
 

Где я ошибаюсь ?

Тушар, вот в чем проблема, которую я вижу в потоках кафки, иногда она съедает эти исключения. но в любом случае, теперь ваша проблема, если она не доходит до другой темы, означает, что это может быть проблема сериализации. также вы делаете это так, как я здесь не вижу.

 var kafkaStreams = new KafkaStreams(sourceStream,  streamsConfiguration);
kafkaStreams.start();