Spring boot kafka messaging. How to simplify dto mapping for handlers?

user_5728095 0
Viewed 2646

I've configured my spring boot project using Kafka. I can receive and publish any string based message.

String message is not the best way for handling. It would be much more useful to have functionality for default converting message from string into object.

Implementing this functionality I've got need to move almost all my Kafka configurations from yml to java (using property). ...producer example

@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, AccountSerializer.class);
    return props;
}

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

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

The code works, but I accepted simplification. In best case I would like to have elegantly configured yml, may be some java changes. But doing it in direct way I'll get additional per 3 beans for configuring each kafkaTemplate and listenerFactory.

Is it possibly simplify the configurations for the future (I'll need more additional Serializer`Deserializer`)? How?

P.S.

I would like configure yml in similar way with this example :

spring:
  kafka:
    consumer:
      group-id: foo
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

app:
  topic:
    foo: foo.t

but I'm not clear how configure here consumers\producers with different (De)Serializers, map ones on specified topic...

apache-kafka spring-boot spring java
2 Answers

It seems I do not have any chances configure for same listener different SERIALIZER|DESERIALIZERs.

But id doesn't mean that my issue has no solution.

I've used inheritance for all my objects and provided an abstraction AbstractEvent. AbstractEvent useless in general but it is used in my solution like enter point for specified SERIALIZER|DESERIALIZER. To get info which object is in context I've used custom headers. org.apache.kafka.common.serialization.Deserializer has no header parameters, but I've implemented my DESERIALIZER basing on ExtendedDeserializer. This way gives me access to headers

via public T deserialize(String topic, Headers headers, byte[] data)

example of my deserializer:

@Slf4j
public class AbstractEventDeserializer<T extends AbstractEvent> implements ExtendedDeserializer<T> {

    private Map<String, Class<T>> mappers = new HashMap<>();

    // default behavior
    @Override
    public T deserialize(String arg0, byte[] devBytes) {
        ObjectMapper mapper = new ObjectMapper();
        T bar = null;
        try {
            bar = (T) mapper.readValue(devBytes, Bar.class);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return bar;
    }

    @Override
    public void close() {
        // TODO Auto-generated method stub
    }

    @Override
    public T deserialize(String topic, Headers headers, byte[] data) {
        log.info("handling...");
        headers.forEach(header -> log.info("   {}: {}", header.key(), getHeaderValueAsString(header)));
        Optional<String> classTypeFromHeader = getClassTypeFromHeader(headers);
        if (classTypeFromHeader.isPresent()) {
            return parseFromJson(data, mappers.get(classTypeFromHeader.get()));
        }
        return deserialize(topic, data);
    }

    private Optional<String> getClassTypeFromHeader(Headers headers) {
        return StreamSupport.stream(headers.headers("X-CLASS-TYPE").spliterator(), false)
                .map(Header::value)
                .map(String::new)
                .findFirst();
    }

    private String getHeaderValueAsString(Header header) {
        return Optional.ofNullable(header.value())
                .map(String::new)
                .orElse(null);
    }

    @Override
    public void configure(Map<String, ?> arg0, boolean arg1) {
        log.info("configuring deserialiser");
        if (arg0.containsKey("mappers")) {
            this.mappers = (Map<String, Class<T>>) arg0.get("mappers");
        }
        arg0.keySet().forEach(key -> log.info("   {}:{}", key, arg0.get(key)));
    }

}

in case you wanna try working solution, please check experimental example.

user_57280950

Spring cloud services allows better configuration for consumers, concurrency, deserialization and less boiler plate code. 

   <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-stream-kafka</artifactId>
    </dependency>

Sample of sink

@SpringBootApplication
@EnableBinding(Sink.class)
public class LoggingConsumerApplication {

public static void main(String[] args) {
    SpringApplication.run(LoggingConsumerApplication.class, args);
}

@StreamListener(Sink.INPUT)
public void handle(Person person) {
    System.out.println("Received: " + person);
}

public static class Person {
    private String name;
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
    public String toString() {
        return this.name;
    }
}
}

Sample configuration: 

spring:
  cloud:
    stream:
      bindings:
        input:
          destination: <your topic>
          group: <your consumer group>
          consumer:
            headerMode: raw
            partitioned: true
            concurrency: 10
      kafka:
        binder:
          brokers: <Comma seperated list of kafka brokers>

More information available here https://cloud.spring.io/spring-cloud-stream/

user_60081560
Related