Repository: LangStream/langstream (language: python)

Issue:
Error when outputting map to streaming topic 
I am using this app configuration:

```
topics:
  - name: "input-topic"
    creation-mode: create-if-not-exists
  - name: "output-topic"
    creation-mode: create-if-not-exists
  - name: "history-topic"
    creation-mode: create-if-not-exists
errors:
    on-failure: skip
pipeline:
  - name: "convert-to-json"
    type: "document-to-json"
    input: "input-topic"
    configuration:
      text-field: "question"
  - name: "ai-chat-completions"
    type: "ai-chat-completions"
    output: "history-topic"
    configuration:
      model: "{{{secrets.open-ai.chat-completions-model}}}" # This needs to be set to the model deployment name, not the base name
      # on the log-topic we add a field with the answer
      completion-field: "value.answer"
      # we are also logging the prompt we sent to the LLM
      log-field: "value.prompt"
      # here we configure the streaming behavior
      # as soon as the LLM answers with a chunk we send it to the answers-topic
      stream-to-topic: "output-topic"
      # on the streaming answer we send the answer as whole message
      # the 'value' syntax is used to refer to the whole value of the message
      
      messages:
        - role: user
          content: "You are a helpful assistant. Below you can find a question from the user. Please try to help them the best way you can.\n\n{{% value.question}}"```

When I send a message to the input topic, it fails with this log:

```
[test-ai-pipeline-document-to-json-1-0] 14:53:35.324 [stats-pipeline-document-to-json-1] INFO  a.l.runtime.agent.AgentRunner -- Records: total 5, working 5, Memory stats: used 40 MB, total 54 MB, free 13 MB, max 121 MB Direct memory 8 MB
[test-ai-pipeline-document-to-json-1-0] 14:53:35.454 [boundedElastic-5] ERROR reactor.core.publisher.Operators -- Operator called default onErrorDropped
[test-ai-pipeline-document-to-json-1-0] reactor.core.Exceptions$ErrorCallbackNotImplemented: java.util.concurrent.CompletionException: java.lang.IllegalArgumentException: Cannot find a serializer for class java.util.HashMap
[test-ai-pipeline-document-to-json-1-0] Caused by: java.util.concurrent.CompletionException: java.lang.IllegalArgumentException: Cannot find a serializer for class java.util.HashMap
[test-ai-pipeline-document-to-json-1-0] 	at java.base/java.util.concurrent.CompletableFuture.reportJoin(CompletableFuture.java:413)
[test-ai-pipeline-document-to-json-1-0] 	at java.base/java.util.concurrent.CompletableFuture.join(CompletableFuture.java:2118)
[test-ai-pipeline-document-to-json-1-0] 	at ai.langstream.ai.agents.GenAIToolKitAgent$TopicStreamingAnswersConsumer.streamAnswerChunk(GenAIToolKitAgent.java:382)
[test-ai-pipeline-document-to-json-1-0] 	at com.datastax.oss.streaming.ai.ChatCompletionsStep$2.consumeChunk(ChatCompletionsStep.java:166)
[test-ai-pipeline-document-to-json-1-0] 	at ai.langstream.ai.agents.services.impl.OpenAICompletionService$ChatCompletionsConsumer.accept(OpenAICompletionService.java:195)
[test-ai-pipeline-document-to-json-1-0] 	at ai.langstream.ai.agents.services.impl.OpenAICompletionService$ChatCompletionsConsumer.accept(OpenAICompletionService.java:132)
[test-ai-pipeline-document-to-json-1-0] 	at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onNext(FluxPeekFuseable.java:196)
[test-ai-pipeline-document-to-json-1-0] 	at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onNext(FluxPeekFuseable.java:210)
[test-ai-pipeline-document-to-json-1-0] 	at reactor.core.publisher.FluxReplay$UnboundedReplayBuffer.replayNormal(FluxReplay.java:618)
[test-ai-pipeline-document-to-json-1-0] 	at reactor.core.publisher.FluxReplay$UnboundedReplayBuffer.replay(FluxReplay.java:709)
[test-ai-pipeline-document-to-json-1-0] 	at reactor.core.publisher.FluxReplay$ReplaySubscriber.onNext(FluxReplay.java:1344)
[test-ai-pipeline-document-to-json-1-0] 	at reactor.core.publisher.FluxConcatMapNoPrefetch$FluxConcatMapNoPrefetchSubscriber.innerNext(FluxConcatMapNoPrefetch.java:258)
[test-ai-pipeline-document-to-json-1-0] 	at reactor.core.publisher.FluxConcatMap$ConcatMapInner.onNext(FluxConcatMap.java:863)
[test-ai-pipeline-document-to-json-1-0] 	at reactor.core.publisher.FluxIterable$IterableSubscription.slowPath(FluxIterable.java:335)
[test-ai-pipeline-document-to-json-1-0] 	at reactor.core.publisher.FluxIterable$IterableSubscription.request(FluxIterable.java:294)
[test-ai-pipeline-document-to-json-1-0] 	at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.set(Operators.java:2341)
[test-ai-pipeline-document-to-json-1-0] 	at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.onSubscribe(Operators.java:2215)
[test-ai-pipeline-document-to-json-1-0] 	at reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:201)
[test-ai-pipeline-document-to-json-1-0] 	at reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:83)
[test-ai-pipeline-document-to-json-1-0] 	at reactor.core.publisher.Flux.subscribe(Flux.java:8671)
[test-ai-pipeline-document-to-json-1-0] 	at reactor.core.publisher.FluxConcatMapNoPrefetch$FluxConcatMapNoPrefetchSubscriber.onNext(FluxConcatMapNoPrefetch.java:206)
[test-ai-pipeline-document-to-json-1-0] 	at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.runAsync(FluxPublishOn.java:440)
[test-ai-pipeline-document-to-json-1-0] 	at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.run(FluxPublishOn.java:527)
[test-ai-pipeline-document-to-json-1-0] 	at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84)
[test-ai-pipeline-document-to-json-1-0] 	at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37)
[test-ai-pipeline-document-to-json-1-0] 	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
[test-ai-pipeline-document-to-json-1-0] 	at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
[test-ai-pipeline-document-to-json-1-0] 	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
[test-ai-pipeline-document-to-json-1-0] 	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
[test-ai-pipeline-document-to-json-1-0] 	at java.base/java.lang.Thread.run(Thread.java:833)
[test-ai-pipeline-document-to-json-1-0] Caused by: java.lang.IllegalArgumentException: Cannot find a serializer for class java.util.HashMap
[test-ai-pipeline-document-to-json-1-0] 	at ai.langstream.kafka.runner.KafkaProducerWrapper.getSerializer(KafkaProducerWrapper.java:237)
[test-ai-pipeline-document-to-json-1-0] 	at ai.langstream.kafka.runner.KafkaProducerWrapper.write(KafkaProducerWrapper.java:181)
[test-ai-pipeline-document-to-json-1-0] 	at ai.langstream.api.runner.topics.TopicConnectionsRuntimeAndLoader$1$3.lambda$write$3(TopicConnectionsRuntimeAndLoader.java:217)
[test-ai-pipeline-document-to-json-1-0] 	at ai.langstream.api.runner.topics.TopicConnectionsRuntimeAndLoader.callNoExceptionWithContextClassloader(TopicConnectionsRuntimeAndLoader.java:62)
[test-ai-pipeline-document-to-json-1-0] 	at ai.langstream.api.runner.topics.TopicConnectionsRuntimeAndLoader$1$3.write(TopicConnectionsRuntimeAndLoader.java:216)
[test-ai-pipeline-document-to-json-1-0] 	... 28 common frames omitted
[test-ai-pipeline-document-to-json-1-0] 14:53:40.324 [stats-pipeline-document-to-json-1] INFO  a.l.runtime.agent.AgentRunner -- Records: total 5, working 5, Memory stats: used 40 MB, total 54 MB, free 13 MB, max 121 MB Direct memory 8 MB```

If I changed this:
```
      completion-field: "value.answer"
```

To this:

```
      completion-field: "value"
```
 It works, but the output is not structured.




Discussion / hints:

Fix patch:
diff --git a/langstream-kafka-runtime/src/main/java/ai/langstream/kafka/runner/KafkaProducerWrapper.java b/langstream-kafka-runtime/src/main/java/ai/langstream/kafka/runner/KafkaProducerWrapper.java
index f5424ad22..d069eec66 100644
--- a/langstream-kafka-runtime/src/main/java/ai/langstream/kafka/runner/KafkaProducerWrapper.java
+++ b/langstream-kafka-runtime/src/main/java/ai/langstream/kafka/runner/KafkaProducerWrapper.java
@@ -15,14 +15,17 @@
  */
 package ai.langstream.kafka.runner;
 
+import static java.util.Map.entry;
 import static org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG;
 import static org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG;
 
 import ai.langstream.api.runner.code.Header;
 import ai.langstream.api.runner.code.Record;
 import ai.langstream.api.runner.topics.TopicProducer;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import io.confluent.kafka.serializers.KafkaAvroSerializer;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -31,6 +34,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.kafka.clients.producer.KafkaProducer;
@@ -51,16 +55,16 @@
 class KafkaProducerWrapper implements TopicProducer {
 
     final Map<Class<?>, Serializer<?>> BASE_SERIALIZERS =
-            Map.of(
-                    String.class, new StringSerializer(),
-                    Boolean.class, new BooleanSerializer(),
-                    Short.class, new ShortSerializer(),
-                    Integer.class, new IntegerSerializer(),
-                    Long.class, new LongSerializer(),
-                    Float.class, new FloatSerializer(),
-                    Double.class, new DoubleSerializer(),
-                    byte[].class, new ByteArraySerializer(),
-                    UUID.class, new UUIDSerializer());
+            Map.ofEntries(
+                    entry(String.class, new StringSerializer()),
+                    entry(Boolean.class, new BooleanSerializer()),
+                    entry(Short.class, new ShortSerializer()),
+                    entry(Integer.class, new IntegerSerializer()),
+                    entry(Long.class, new LongSerializer()),
+                    entry(Float.class, new FloatSerializer()),
+                    entry(Double.class, new DoubleSerializer()),
+                    entry(byte[].class, new ByteArraySerializer()),
+                    entry(UUID.class, new UUIDSerializer()));
 
     final Map<Class<?>, Serializer<?>> keySerializers = new ConcurrentHashMap<>(BASE_SERIALIZERS);
     final Map<Class<?>, Serializer<?>> valueSerializers = new ConcurrentHashMap<>(BASE_SERIALIZERS);
@@ -226,21 +230,41 @@ public synchronized CompletableFuture<?> write(Record r) {
 
     private Serializer<?> getSerializer(
             Class<?> r, Map<Class<?>, Serializer<?>> serializerMap, Boolean isKey) {
-        Serializer<?> result = serializerMap.get(r);
-        if (result == null) {
-            if (GenericRecord.class.isAssignableFrom(r) && isKey != null) { // no AVRO in headers
-                KafkaAvroSerializer kafkaAvroSerializer = new KafkaAvroSerializer();
-                kafkaAvroSerializer.configure(copy, isKey);
-                serializerMap.put(r, kafkaAvroSerializer);
-                return kafkaAvroSerializer;
-            }
-            throw new IllegalArgumentException("Cannot find a serializer for " + r);
-        }
-        return result;
+        return serializerMap.computeIfAbsent(
+                r,
+                k -> {
+                    if (GenericRecord.class.isAssignableFrom(k)
+                            && isKey != null) { // no AVRO in headers
+                        KafkaAvroSerializer kafkaAvroSerializer = new KafkaAvroSerializer();
+                        kafkaAvroSerializer.configure(copy, isKey);
+                        serializerMap.put(r, kafkaAvroSerializer);
+                        return kafkaAvroSerializer;
+                    }
+                    if (Map.class.isAssignableFrom(k)) {
+                        return new ObjectToJsonSerializer();
+                    }
+                    if (Collection.class.isAssignableFrom(k)) {
+                        return new ObjectToJsonSerializer();
+                    }
+                    throw new IllegalArgumentException("Cannot find a serializer for " + r);
+                });
     }
 
     @Override
     public String toString() {
         return "KafkaProducerWrapper{" + "topicName='" + topicName + '\'' + '}';
     }
+
+    private static class ObjectToJsonSerializer implements Serializer<Object> {
+        private static final ObjectMapper MAPPER = new ObjectMapper();
+
+        @Override
+        public void configure(Map<String, ?> configs, boolean isKey) {}
+
+        @Override
+        @SneakyThrows
+        public byte[] serialize(String topic, Object data) {
+            return MAPPER.writeValueAsBytes(data);
+        }
+    }
 }


Test patch:
diff --git a/langstream-kafka-runtime/src/test/java/ai/langstream/kafka/KafkaConsumerTest.java b/langstream-kafka-runtime/src/test/java/ai/langstream/kafka/KafkaConsumerTest.java
index 32f182ff8..e45add6ee 100644
--- a/langstream-kafka-runtime/src/test/java/ai/langstream/kafka/KafkaConsumerTest.java
+++ b/langstream-kafka-runtime/src/test/java/ai/langstream/kafka/KafkaConsumerTest.java
@@ -481,6 +481,8 @@ public void testMultipleSchemas() throws Exception {
                                         new SimpleRecord.SimpleHeader("h2", "bar")))
                         .join();
 
+                producer.write(generateRecord(List.of("one", "two"), Map.of("k", "v"))).join();
+
                 try (KafkaConsumerWrapper consumer =
                         (KafkaConsumerWrapper)
                                 runtime.createConsumer(