idealo/deckard

Easy-to-use Spring Kafka Producers

deckard's Issues

Add support for kafka properties per producer

Currently kafka properties are applied to all producers even if they have different bootstrap servers leading the application to fail in startup.

Just as an example if one cluster has SSL configured and another not, the application will try to connect with all producers for all the bootstrap servers with this config, leading to SSL handshake exceptions.

Add support to KafkaAvroSerializer

When trying to configure a producer with avro serializer, deckard fails to fetch the avro configuration because they are custom properties resulting in a null pointer exception.

Properties used in application.yml where:

spring:
  kafka:
    properties:
      specific.avro.reader: true
      schema.registry.url: avro_schema_url

Stacktrace:
The configuration 'schema.registry.url' was supplied but isn't a known config. 2020-09-11 15:12:12.283 WARN ProducerConfig [http-nio-8080-exec-1] [traceId=62babe4eb002a341, spanId=62babe4eb002a341, spanExportable=false, X-Span-Export=false, X-B3-SpanId=62babe4eb002a341, X-B3-TraceId=62babe4eb002a341] - The configuration 'specific.avro.reader' was supplied but isn't a known config. 2020-09-11 15:12:12.585 ERROR [dispatcherServlet] [http-nio-8080-exec-1] [] - Servlet.service() for servlet [dispatcherServlet] in context with path [] threw exception [Request processing failed; nested exception is org.apache.kafka.common.errors.SerializationException: Error serializing Avro message] with root cause java.lang.NullPointerException: null at io.confluent.kafka.serializers.AbstractKafkaAvroSerializer.serializeImpl(AbstractKafkaAvroSerializer.java:74) at io.confluent.kafka.serializers.KafkaAvroSerializer.serialize(KafkaAvroSerializer.java:53) at org.apache.kafka.common.serialization.Serializer.serialize(Serializer.java:62) at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:902) at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:862) at org.springframework.kafka.core.DefaultKafkaProducerFactory$CloseSafeProducer.send(DefaultKafkaProducerFactory.java:781) at org.springframework.kafka.core.KafkaTemplate.doSend(KafkaTemplate.java:562) at org.springframework.kafka.core.KafkaTemplate.send(KafkaTemplate.java:369) at de.idealo.deckard.producer.Producer.send(Producer.java:19) at de.idealo.deckard.proxy.ProducerInvocationHandler.invoke(ProducerInvocationHandler.java:22) at com.sun.proxy.$Proxy112.send(Unknown Source) at de.idealo.shopping.barcamp.services.MyService.produceMessage(MyService.java:33) at de.idealo.shopping.barcamp.controllers.SampleController.publishMsg(SampleController.java:19) at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.base/java.lang.reflect.Method.invoke(Method.java:566) at org.springframework.web.method.support.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:190) at org.springframework.web.method.support.InvocableHandlerMethod.invokeForRequest(InvocableHandlerMethod.java:138) at org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(ServletInvocableHandlerMethod.java:105) at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.invokeHandlerMethod(RequestMappingHandlerAdapter.java:878) at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.handleInternal(RequestMappingHandlerAdapter.java:792) at org.springframework.web.servlet.mvc.method.AbstractHandlerMethodAdapter.handle(AbstractHandlerMethodAdapter.java:87) at org.springframework.web.servlet.DispatcherServlet.doDispatch(DispatcherServlet.java:1040) at org.springframework.web.servlet.DispatcherServlet.doService(DispatcherServlet.java:943) at org.springframework.web.servlet.FrameworkServlet.processRequest(FrameworkServlet.java:1006) at org.springframework.web.servlet.FrameworkServlet.doGet(FrameworkServlet.java:898) at javax.servlet.http.HttpServlet.service(HttpServlet.java:626) at org.springframework.web.servlet.FrameworkServlet.service(FrameworkServlet.java:883) at javax.servlet.http.HttpServlet.service(HttpServlet.java:733) at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:231) at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166) at org.apache.tomcat.websocket.server.WsFilter.doFilter(WsFilter.java:53) at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193) at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166) at brave.servlet.TracingFilter.doFilter(TracingFilter.java:68) at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193) at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166) at org.springframework.web.filter.RequestContextFilter.doFilterInternal(RequestContextFilter.java:100) at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119) at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193) at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166) at org.springframework.web.filter.FormContentFilter.doFilterInternal(FormContentFilter.java:93) at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119) at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193) at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166) at brave.servlet.TracingFilter.doFilter(TracingFilter.java:87) at org.springframework.cloud.sleuth.instrument.web.LazyTracingFilter.doFilter(TraceWebServletAutoConfiguration.java:139) at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193) at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166) at org.springframework.boot.actuate.metrics.web.servlet.WebMvcMetricsFilter.doFilterInternal(WebMvcMetricsFilter.java:93) at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119) at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193) at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166) at org.springframework.web.filter.CharacterEncodingFilter.doFilterInternal(CharacterEncodingFilter.java:201) at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119) at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193) at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166) at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:202) at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:96) at org.apache.catalina.authenticator.AuthenticatorBase.invoke(AuthenticatorBase.java:541) at de.idealo.common.springboot.accesslog.SleuthValve.invoke(SleuthValve.java:36) at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:139) at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:92) at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:74) at net.rakugakibox.spring.boot.logback.access.tomcat.LogbackAccessTomcatValve.invoke(LogbackAccessTomcatValve.java:91) at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:343) at org.apache.coyote.http11.Http11Processor.service(Http11Processor.java:373) at org.apache.coyote.AbstractProcessorLight.process(AbstractProcessorLight.java:65) at org.apache.coyote.AbstractProtocol$ConnectionHandler.process(AbstractProtocol.java:868) at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:1589) at org.apache.tomcat.util.net.SocketProcessorBase.run(SocketProcessorBase.java:49) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61) at java.base/java.lang.Thread.run(Thread.java:834) ACCESS 0:0:0:0:0:0:0:1 11/Sep/2020:15:12:12 +0200 'GET / HTTP/1.1' 500 forwarded-for:- userid:- X-B3-TraceId: 62babe4eb002a341

DecryptingDeserializer does not support JsonDeserializer (ExtendedDeserializer)

The DecryptingDeserializer does not support JsonDeserializer as embedded deserializer. This is most likely caused by the JsonDeserializer being an implementation of ExtendedDeserializer rather than directly Deserializer.

    @Test
    public void shouldDecryptOnJsonDeserialize() {
        EncryptingSerializer<MyPayload> encryptingSerializer = new EncryptingSerializer<>(PASS, SALT, new JsonSerializer<>());
        DecryptingDeserializer<MyPayload> decryptingDeserializer = new DecryptingDeserializer<>(PASS, SALT, new JsonDeserializer<>());
        byte[] encryptedData = encryptingSerializer.serialize("my-topic", new MyPayload("my-data"));
        MyPayload decryptedData = decryptingDeserializer.deserialize("my-topic", encryptedData);
        Assertions.assertThat(decryptedData).isEqualTo(new MyPayload("my-data"));
    }

causes java.lang.IllegalArgumentException: Unrecognized Type: [null]