Spring Cloud Stream отправляет данные темам Kafka - PullRequest
Новая
       8

Spring Cloud Stream отправляет данные темам Kafka

0 голосов
/ 17 ноября 2018

У меня есть приложение SCS Sink.Я пытаюсь отправить сообщение MessageBuilder темам kafka с помощью Spring Kafka

ALso,

Я пытаюсь выполнить условную пересылку с использованием Spring Kafka здесь.Есть ли способ сделать это в соответствии с приведенным ниже кодом, используя SCS (например, @SendTo и т. Д.)?В идеале я хотел бы использовать процессор, но не раковину.Просто непонятно по условной пересылке.

Спасибо за ваше время.

demoSink: 

        @EnableBinding(Sink.class)
    public class SampleSink {

        @ServiceActivator(inputChannel = Sink.INPUT)
        public void processor(Message<?> message) {

            @Autowired
            private Sender sender;

            Message<String> message1 = (Message<String>) MessageBuilder.fromMessage(message)
                                        .setHeader("header1", "header1")
                                        .copyHeadersIfAbsent(message.getHeaders())
                                        .build();
    if(someObj.getSomething()==something){
        sender.sendt1(message1);
}
    else{

    sender.sendt2(message)
}

    }

Sender.java 

   @Service
    public class Sender {

        private static final Logger LOG = LoggerFactory.getLogger(Sender.class);

        @Autowired
        private KafkaTemplate<String, String> kafkaTemplate;

     @Val

ue("${demo.topic}")
        private String topic1;

    public void sendt1(Message<?> msg){

           Message<String> t1 = (Message<String>) MessageBuilder.fromMessage(msg)
                   .copyHeadersIfAbsent(msg.getHeaders())
                    .setHeader(KafkaHeaders.TOPIC, topic1)
                    .build();

            LOG.info("sending message topic1");
            kafkaTemplate.send(t1);
        }


    public void sendt2(Message<?> msg){

           Message<String> t2 = (Message<String>) MessageBuilder.fromMessage(msg)
                   .copyHeadersIfAbsent(msg.getHeaders())
                    .setHeader(KafkaHeaders.TOPIC, topic2)
                    .build();

            LOG.info("sending message topic2");
            kafkaTemplate.send(t2);
        }
    }

config: 

@Configuration
public class SenderConfig {

    @Value("${spring.cloud.stream.kafka.binder.brokers}")
    private String bootstrapServers;

    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return props;
    }

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

}

Я получаю следующую ошибку: - 

    Caused by:
 org.apache.kafka.common.errors.SerializationException: Can't convert value of class [B to class org.apache.kafka.common.serialization.StringSerializer
 specified in value.serializer
Caused by: java.lang.ClassCastException: 
[B cannot be cast to java.lang.String

1 Ответ

0 голосов
/ 17 ноября 2018

spring пытается преобразовать ByteArray в String Object, если вы пытаетесь только пересылать сообщения, вы можете сделать это следующим образом: изменить на: 

Message<?> message1 = MessageBuilder.fromMessage(message)
                                    .setHeader("header1", "header1")
                                    .copyHeadersIfAbsent(message.getHeaders())
                                    .build();

и вместо этого в вашем SenderConfig использовать ByteArraySerializer.classдля ключа и значения.

...