yujunjun/datahub
2
0
Fork 0
mirror of https://github.com/datahub-project/datahub.git synced 2025-08-26 18:15:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(metadata-service): stop overriding kafka registry props with empty values (#4604)

Browse Source 
Co-authored-by: jsotelo <javier.sotelo@viasat.com>
This commit is contained in:
Javier Sotelo 2022-04-11 22:02:47 -07:00 committed by GitHub
parent e7c5eb357c
commit fa4cc3ad31
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 12 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
metadata-service/factories/src/main/java/com/linkedin/gms/factory/kafka/DataHubKafkaProducerFactory.java
View File

@ -65,7 +65,12 @@ public class DataHubKafkaProducerFactory {
Map<String, Object> props = properties.buildProducerProperties();
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, schemaRegistryConfig.getSerializer().getName());
props.putAll(schemaRegistryConfig.getProperties());
// Override KafkaProperties with SchemaRegistryConfig only for non-empty values
schemaRegistryConfig.getProperties().entrySet()
.stream()
.filter(entry -> entry.getValue() != null && !entry.getValue().toString().isEmpty())
.forEach(entry -> props.put(entry.getKey(), entry.getValue()));
return new KafkaProducer<>(props);
}

7
metadata-service/factories/src/main/java/com/linkedin/gms/factory/kafka/KafkaEventConsumerFactory.java
View File

@ -76,7 +76,12 @@ public class KafkaEventConsumerFactory {
consumerProps.setValueDeserializer(schemaRegistryConfig.getDeserializer());
Map<String, Object> props = properties.buildConsumerProperties();
props.putAll(schemaRegistryConfig.getProperties());
// Override KafkaProperties with SchemaRegistryConfig only for non-empty values
schemaRegistryConfig.getProperties().entrySet()
.stream()
.filter(entry -> entry.getValue() != null && !entry.getValue().toString().isEmpty())
.forEach(entry -> props.put(entry.getKey(), entry.getValue()));
ConcurrentKafkaListenerContainerFactory<String, GenericRecord> factory =
new ConcurrentKafkaListenerContainerFactory<>();