Disable __TypeId__ header for AAI event publishing

- with the refactorings in AAI-3931 and AAI-4014 message producing was changed
  to serialize <String,NotificationEvent> instead of <String,String> to be more expressive
- when doing so, spring will add a __TypeId__ header to the message to inform consumers
  about the type of the message
- spring consumers will validate this annotation and will
    - throw an exception when the package (org.onap.aai.[...].NotificationEvent) is not among the trusted packages
    - throw an exception if the class is not on the classpath and the app thus does not know how to deserialize it
- since this is a breaking change for the consumers and it would just cause implementation effort, the header is removed to restore the old behaviour

Issue-ID: AAI-4074
Change-Id: Ibf83c235739a71b64471ee85d0b5e559082c271e
Signed-off-by: Fiete Ostkamp <Fiete.Ostkamp@telekom.de>

diff --git a/aai-core/src/main/java/org/onap/aai/config/KafkaConfig.java b/aai-core/src/main/java/org/onap/aai/config/KafkaConfig.java
index 7a81d9d..69a6814 100644 (file)
--- a/aai-core/src/main/java/org/onap/aai/config/KafkaConfig.java
+++ b/aai-core/src/main/java/org/onap/aai/config/KafkaConfig.java
@@ -61,6 +61,9 @@ public class KafkaConfig {
     @Value("${spring.kafka.producer.retries:3}")
     private String retries;
 
+    @Value("${spring.kafka.producer.maxInFlightConnections:10}")
+    private String maxInFlightConnections;
+
     private Map<String, Object> buildKafkaProperties() throws Exception {
         Map<String, Object> props = new HashMap<>();
         if (bootstrapServers == null) {
@@ -71,8 +74,9 @@ public class KafkaConfig {
         }
         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        props.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false);
         props.put(ProducerConfig.RETRIES_CONFIG, retries);
-        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5");
+        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, maxInFlightConnections);
 
         if (saslJaasConfig == null) {
             logger.info("Not using any authentication for kafka interaction");