Create v3

bakdata · Apr 5, 2024 · 042757e · 042757e
1 parent 220083a
commit 042757e
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 6 deletions.
diff --git a/streams-bootstrap/src/main/java/com/bakdata/kafka/ConfiguredProducerApp.java b/streams-bootstrap/src/main/java/com/bakdata/kafka/ConfiguredProducerApp.java
@@ -73,10 +73,11 @@ private static Map<String, Object> createKafkaProperties(final KafkaEndpointConf
      *          Configs provided by {@link ProducerApp#createKafkaProperties()}
      *      </li>
      *      <li>
-     *          Configs provided by {@link ProducerAppConfiguration#getKafkaConfig()}
+     *          Configs provided via environment variables (see
+     *          {@link EnvironmentKafkaConfigParser#parseVariables(Map)})
      *      </li>
      *      <li>
-     *          Environment variables {see @{link {@link EnvironmentKafkaConfigParser#parseVariables(Map)}}}
+     *          Configs provided by {@link ProducerAppConfiguration#getKafkaConfig()}
      *      </li>
      *      <li>
      *          Configs provided by {@link KafkaEndpointConfig#createKafkaProperties()}
@@ -88,8 +89,8 @@ private static Map<String, Object> createKafkaProperties(final KafkaEndpointConf
     public Map<String, Object> getKafkaProperties(final KafkaEndpointConfig endpointConfig) {
         final Map<String, Object> kafkaConfig = createKafkaProperties(endpointConfig);
         kafkaConfig.putAll(this.app.createKafkaProperties());
-        kafkaConfig.putAll(this.configuration.getKafkaConfig());
         kafkaConfig.putAll(EnvironmentKafkaConfigParser.parseVariables(System.getenv()));
+        kafkaConfig.putAll(this.configuration.getKafkaConfig());
         kafkaConfig.putAll(endpointConfig.createKafkaProperties());
         return Collections.unmodifiableMap(kafkaConfig);
     }

diff --git a/streams-bootstrap/src/main/java/com/bakdata/kafka/ConfiguredStreamsApp.java b/streams-bootstrap/src/main/java/com/bakdata/kafka/ConfiguredStreamsApp.java
@@ -74,10 +74,11 @@ private static Map<String, Object> createKafkaProperties(final KafkaEndpointConf
      *          Configs provided by {@link StreamsApp#createKafkaProperties(StreamsConfigurationOptions)}
      *      </li>
      *      <li>
-     *          Configs provided by {@link StreamsAppConfiguration#getKafkaConfig()}
+     *          Configs provided via environment variables (see
+     *          {@link EnvironmentKafkaConfigParser#parseVariables(Map)})
      *      </li>
      *      <li>
-     *          Environment variables {see @{link {@link EnvironmentKafkaConfigParser#parseVariables(Map)}}}
+     *          Configs provided by {@link StreamsAppConfiguration#getKafkaConfig()}
      *      </li>
      *      <li>
      *          Configs provided by {@link KafkaEndpointConfig#createKafkaProperties()}
@@ -93,8 +94,8 @@ private static Map<String, Object> createKafkaProperties(final KafkaEndpointConf
     public Map<String, Object> getKafkaProperties(final KafkaEndpointConfig endpointConfig) {
         final Map<String, Object> kafkaConfig = createKafkaProperties(endpointConfig);
         kafkaConfig.putAll(this.app.createKafkaProperties(this.configuration.getOptions()));
-        kafkaConfig.putAll(this.configuration.getKafkaConfig());
         kafkaConfig.putAll(EnvironmentKafkaConfigParser.parseVariables(System.getenv()));
+        kafkaConfig.putAll(this.configuration.getKafkaConfig());
         kafkaConfig.putAll(endpointConfig.createKafkaProperties());
         kafkaConfig.put(StreamsConfig.APPLICATION_ID_CONFIG, this.app.getUniqueAppId(this.getTopics()));
         return Collections.unmodifiableMap(kafkaConfig);