Change in record formatter to add namespace to topic in ad-hoc tests …

…(didn't work previously when naming strategy was set)
TouK · Nov 27, 2024 · f7e4dd4 · f7e4dd4
1 parent fae79e6
commit f7e4dd4
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 7 deletions.
diff --git a/...n/scala/pl/touk/nussknacker/engine/schemedkafka/schemaregistry/SchemaRegistryClient.scala b/...n/scala/pl/touk/nussknacker/engine/schemedkafka/schemaregistry/SchemaRegistryClient.scala
@@ -3,11 +3,7 @@ package pl.touk.nussknacker.engine.schemedkafka.schemaregistry
 import cats.data.Validated
 import io.confluent.kafka.schemaregistry.ParsedSchema
 import pl.touk.nussknacker.engine.kafka.{KafkaConfig, UnspecializedTopicName}
-import pl.touk.nussknacker.engine.schemedkafka.{
-  TopicSelectionStrategy,
-  TopicsMatchingPatternWithExistingSubjectsSelectionStrategy,
-  TopicsWithExistingSubjectSelectionStrategy
-}
+import pl.touk.nussknacker.engine.schemedkafka.TopicsWithExistingSubjectSelectionStrategy
 
 trait SchemaRegistryClient extends Serializable {
 

diff --git a/...ker/engine/schemedkafka/schemaregistry/formatter/AbstractSchemaBasedRecordFormatter.scala b/...ker/engine/schemedkafka/schemaregistry/formatter/AbstractSchemaBasedRecordFormatter.scala
@@ -8,7 +8,6 @@ import pl.touk.nussknacker.engine.api.process.TopicName
 import pl.touk.nussknacker.engine.api.test.TestRecord
 import pl.touk.nussknacker.engine.kafka.consumerrecord.SerializableConsumerRecord
 import pl.touk.nussknacker.engine.kafka.{KafkaConfig, RecordFormatter, serialization}
-import pl.touk.nussknacker.engine.schemedkafka.TopicsWithExistingSubjectSelectionStrategy
 import pl.touk.nussknacker.engine.schemedkafka.schemaregistry.{
   ContentTypes,
   ContentTypesSchemas,
@@ -143,7 +142,7 @@ abstract class AbstractSchemaBasedRecordFormatter[K: ClassTag, V: ClassTag] exte
 
     }
 
-    record.consumerRecord.toKafkaConsumerRecord(topic, serializeKeyValue)
+    record.consumerRecord.copy(topic = Some(topic.name)).toKafkaConsumerRecord(topic, serializeKeyValue)
   }
 
   protected def readRecordKeyMessage(