Change dataflow application id to unique (#4569)

SeldonIO · Jan 11, 2023 · 0bd289c · 0bd289c
1 parent 37255bd
commit 0bd289c
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 2 deletions.
diff --git a/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/kafka/Configuration.kt b/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/kafka/Configuration.kt
@@ -105,7 +105,7 @@ fun getKafkaProperties(params: KafkaStreamsParams): KafkaProperties {
         this[StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG] = 0
         this[StreamsConfig.COMMIT_INTERVAL_MS_CONFIG] = 1
 
-        this[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = "earliest"
+        this[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = "latest"
         this[ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG] = KAFKA_MAX_MESSAGE_BYTES
         this[ConsumerConfig.FETCH_MAX_BYTES_CONFIG] = KAFKA_MAX_MESSAGE_BYTES
         this[ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG] = 60000

diff --git a/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/kafka/Pipeline.kt b/scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/kafka/Pipeline.kt
@@ -123,7 +123,7 @@ class Pipeline(
         ): KafkaProperties {
             return kafkaProperties
                 .withAppId(
-                    HashUtils.hashIfLong(metadata.name),
+                    HashUtils.hashIfLong(metadata.id),
                 )
                 .withStreamThreads(
                     getNumThreadsFor(numSteps),