feat: SerDe in Source and Sink Avro Support

consumer-config.yaml

@@ -0,0 +1,20 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: consumer-config
+data:
+  application.yml: |
+    spring:
+      pulsar:
+        client: # see here for all configurations: https://pulsar.apache.org/reference/#/4.0.x/client/client-configuration-client
+          clientConfig:
+            serviceUrl: "pulsar://host.docker.internal:6650"
+        consumer:
+          enabled: true
+          consumerConfig: # see here for all configurations: https://pulsar.apache.org/reference/#/4.0.x/client/client-configuration-producer
+            topicNames: "tester-tester" 
+        admin: 
+          adminConfig: # Accepts the same key-value pair configurations as pulsar client: https://pulsar.apache.org/reference/#/4.0.x/client/client-configuration-client
+            serviceUrl: "http://host.docker.internal:8080"
+
+

consumer-pipeline.yaml

@@ -0,0 +1,43 @@
+apiVersion: numaflow.numaproj.io/v1alpha1
+kind: Pipeline
+metadata:
+  name: consumer-pipeline
+spec:
+  limits: 
+    readBatchSize: 1 # Change if you want a different batch size
+  vertices:
+    - name: in
+      scale:
+        min: 1
+      volumes: 
+        - name: pulsar-config-volume
+          configMap:
+            name: consumer-config
+            items:
+              - key: application.yml
+                path: application.yml
+      source:
+        udsource: 
+          container:
+            image: apache-pulsar-java:v0.3.0
+            args: [ "--spring.config.location=file:/conf/application.yml" ]
+            imagePullPolicy: Never
+            volumeMounts:
+              - name: pulsar-config-volume
+                mountPath: /conf
+    - name: p1
+      scale:
+        min: 1
+      udf:
+        builtin:
+          name: cat
+    - name: out
+      scale:
+        min: 1
+      sink:
+        log: {}
+  edges:
+    - from: in
+      to: p1
+    - from: p1
+      to: out

docker-compose.yml

@@ -1,4 +1,3 @@
-version: '3.5'
 
 services:
   pulsar:

pom.xml

@@ -35,6 +35,16 @@
             <artifactId>spring-boot-starter-test</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.avro</groupId>
+            <artifactId>avro</artifactId>
+            <version>1.11.0</version>
+        </dependency>
+        <dependency>
+            <groupId>org.json</groupId>
+            <artifactId>json</artifactId>
+            <version>20231013</version>
+        </dependency>
     </dependencies>
 
     <build>

producer-config.yaml

@@ -0,0 +1,18 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: producer-config
+data:
+  application.yml: |
+    spring:
+      pulsar:
+        client: # see here for all configurations: https://pulsar.apache.org/reference/#/4.0.x/client/client-configuration-client
+          clientConfig:
+            serviceUrl: "pulsar://host.docker.internal:6650"
+        producer:
+          enabled: true
+          producerConfig: # see here for all configurations: https://pulsar.apache.org/reference/#/4.0.x/client/client-configuration-producer
+            topicName: "tester-tester" 
+            sendTimeoutMs: 2000
+
+

producer-pipeline.yaml

@@ -0,0 +1,44 @@
+apiVersion: numaflow.numaproj.io/v1alpha1
+kind: Pipeline
+metadata:
+  name: producer-pipeline
+spec:
+  vertices:
+    - name: in
+      scale:
+        min: 1
+      source:
+        generator:
+          rpu: 1
+          duration: 1s
+          msgSize: 10
+    - name: p1
+      scale:
+        min: 1
+      udf:
+        builtin:
+          name: cat
+    - name: out
+      scale:
+        min: 1
+      volumes: # Shared between containers that are part of the same pod, useful for sharing configurations
+        - name: pulsar-config-volume
+          configMap:
+            name: producer-config
+            items:
+              - key: application.yml
+                path: application.yml
+      sink:
+        udsink:
+          container:
+            image: apache-pulsar-java:v0.3.0 # TO DO: Replace with quay.io link
+            args: [ "--spring.config.location=file:/conf/application.yml" ] # Use external configuration file 
+            imagePullPolicy: Never
+            volumeMounts:
+              - name: pulsar-config-volume
+                mountPath: /conf
+  edges:
+    - from: in
+      to: p1
+    - from: p1
+      to: out

src/main/java/io/numaproj/pulsar/config/producer/PulsarProducerConfig.java

@@ -8,35 +8,51 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.env.Environment;
+import org.springframework.core.io.ClassPathResource;
 
 import lombok.extern.slf4j.Slf4j;
 
 import java.util.Map;
 import java.util.UUID;
+import java.util.Optional;
 
 @Slf4j
 @Configuration
 public class PulsarProducerConfig {
 
-    @Autowired
-    private Environment env;
-
-    @Bean
-    @ConditionalOnProperty(prefix = "spring.pulsar.producer", name = "enabled", havingValue = "true", matchIfMissing = false)
-    public Producer<byte[]> pulsarProducer(PulsarClient pulsarClient, PulsarProducerProperties pulsarProducerProperties)
-            throws Exception {
-        String podName = env.getProperty("NUMAFLOW_POD", "pod-" + UUID.randomUUID());
-        String producerName = "producerName";
-
-        Map<String, Object> producerConfig = pulsarProducerProperties.getProducerConfig();
-        if (producerConfig.containsKey(producerName)) {
-            log.warn("User configured a 'producerName' in the config, but this can cause errors if multiple pods spin "
-                    + "up with the same name. Overriding with '{}'", podName);
+        @Autowired
+        private Environment env;
+
+        @Bean
+        @ConditionalOnProperty(prefix = "spring.pulsar.producer", name = "enabled", havingValue = "true", matchIfMissing = false)
+        public Producer<byte[]> pulsarProducer(PulsarClient pulsarClient,
+                        PulsarProducerProperties pulsarProducerProperties)
+                        throws Exception {
+                String podName = env.getProperty("NUMAFLOW_POD", "pod-" + UUID.randomUUID());
+                String producerName = "producerName";
+
+                Map<String, Object> producerConfig = pulsarProducerProperties.getProducerConfig();
+                if (producerConfig.containsKey(producerName)) {
+                        log.warn("User configured a 'producerName' in the config, but this can cause errors if multiple pods spin "
+                                        + "up with the same name. Overriding with '{}'", podName);
+                }
+                producerConfig.put(producerName, podName);
+
+                // Optionally load schema for client-side validation if schema file exists
+                try {
+                        String schemaStr = new String(
+                                        new ClassPathResource("schema.avsc").getInputStream().readAllBytes());
+                        org.apache.avro.Schema avroSchema = new org.apache.avro.Schema.Parser().parse(schemaStr);
+                        log.info("Found AVRO schema for client-side validation: {}", avroSchema.toString(true));
+                        pulsarProducerProperties.setAvroSchema(Optional.of(avroSchema));
+                } catch (Exception e) {
+                        log.info("No schema.avsc found or error loading schema. Client-side validation will be disabled.");
+                        pulsarProducerProperties.setAvroSchema(Optional.empty());
+                }
+
+                // Create producer with byte[] schema for maximum flexibility
+                return pulsarClient.newProducer(Schema.BYTES)
+                                .loadConf(producerConfig)
+                                .create();
         }
-        producerConfig.put(producerName, podName);
-
-        return pulsarClient.newProducer(Schema.BYTES)
-                .loadConf(producerConfig)
-                .create();
-    }
 }

src/main/java/io/numaproj/pulsar/config/producer/PulsarProducerProperties.java

@@ -2,17 +2,19 @@
 
 import lombok.Getter;
 import lombok.Setter;
-
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.context.annotation.Configuration;
+import org.apache.avro.Schema;
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
 
 @Getter
 @Setter
 @Configuration
 @ConfigurationProperties(prefix = "spring.pulsar.producer")
 public class PulsarProducerProperties {
     private Map<String, Object> producerConfig = new HashMap<>(); // Default to an empty map
+    private Optional<Schema> avroSchema = Optional.empty(); // Optional schema for client-side validation
 }