Develop Data Transforms

import (
	"encoding/json"
	"github.com/redpanda-data/redpanda/src/transform-sdk/go/transform"
)

func main() {
	transform.OnRecordWritten(filterValidJson)
}

func filterValidJson(event transform.WriteEvent, writer transform.RecordWriter) error {
	if json.Valid(event.Record().Value) {
		return writer.Write(event.Record())
	}
	// Send invalid records to separate topic
	return writer.Write(event.Record(), transform.ToTopic("invalid-json"))
}

use anyhow::Result;
use redpanda_transform_sdk::*;

fn main() {
	on_record_written(filter_valid_json);
}

fn filter_valid_json(event: WriteEvent, writer: &mut RecordWriter) -> Result<()> {
	let value = event.record.value().unwrap_or_default();
	if serde_json::from_slice::<serde_json::Value>(value).is_ok() {
		writer.write(event.record)?;
	} else {
		// Send invalid records to separate topic
		writer.write_with_options(event.record, WriteOptions::to_topic("invalid-json"))?;
	}
	Ok(())
}

package main

import (
    "encoding/binary"
    "encoding/json"
    "log"
    "github.com/redpanda-data/redpanda/src/transform-sdk/go/transform"
    "github.com/redpanda-data/redpanda/src/transform-sdk/go/transform/sr"
)

// Input message structure with array of updates
type BatchMessage struct {
    Updates []TableUpdate `json:"updates"`
}

// Individual table update with routing field
type TableUpdate struct {
    Table string          `json:"table"` // Routing field - determines output topic
    Data  json.RawMessage `json:"data"`  // The actual data to write
}

// Schema IDs for each output topic, registered dynamically at startup
var schemaIDs = make(map[string]int)

func main() {
    // Create Schema Registry client
    client := sr.NewClient()

    // Define schemas for each output topic
    schemas := map[string]string{
        "orders":    `{"type":"record","name":"Order","fields":[{"name":"order_id","type":"string"},{"name":"amount","type":"double"}]}`,
        "inventory": `{"type":"record","name":"Inventory","fields":[{"name":"product_id","type":"string"},{"name":"quantity","type":"int"}]}`,
        "customers": `{"type":"record","name":"Customer","fields":[{"name":"customer_id","type":"string"},{"name":"name","type":"string"}]}`,
    }

    // Register schemas and store their IDs
    for topic, schemaStr := range schemas {
        subject := topic + "-value"
        schema := sr.Schema{
            Schema: schemaStr,
            Type:   sr.TypeAvro,
        }

        result, err := client.CreateSchema(subject, schema)
        if err != nil {
            log.Fatalf("Failed to register schema for %s: %v", topic, err)
        }

        schemaIDs[topic] = result.ID
        log.Printf("Registered schema for %s with ID %d", topic, result.ID)
    }

    log.Printf("Starting fanout transform with schema IDs: %v", schemaIDs)
    transform.OnRecordWritten(routeUpdates)
}

func routeUpdates(event transform.WriteEvent, writer transform.RecordWriter) error {
    var batch BatchMessage
    if err := json.Unmarshal(event.Record().Value, &batch); err != nil {
        log.Printf("Failed to parse batch message: %v", err)
        return nil // Skip invalid records
    }

    // Process each update in the batch
    for i, update := range batch.Updates {
        schemaID, exists := schemaIDs[update.Table]
        if !exists {
            log.Printf("Unknown table in update %d: %s", i, update.Table)
            continue
        }

        if err := writeUpdate(update, schemaID, writer, event); err != nil {
            log.Printf("Failed to write update %d to %s: %v", i, update.Table, err)
        }
    }

    return nil
}

func writeUpdate(update TableUpdate, schemaID int, writer transform.RecordWriter, event transform.WriteEvent) error {
    // Create Schema Registry wire format: [magic_byte, schema_id (4 bytes BE), data...]
    value := make([]byte, 5)
    value[0] = 0 // magic byte
    binary.BigEndian.PutUint32(value[1:5], uint32(schemaID))
    value = append(value, update.Data...)

    record := transform.Record{
        Key:   event.Record().Key,
        Value: value,
    }

    return writer.Write(record, transform.ToTopic(update.Table))
}

use redpanda_transform_sdk::*;
use redpanda_transform_sdk_sr::{SchemaRegistryClient, Schema, SchemaFormat};
use serde::Deserialize;
use std::collections::HashMap;
use std::error::Error;
use std::sync::OnceLock;
use log::{info, error};

#[derive(Deserialize)]
struct BatchMessage {
    updates: Vec<TableUpdate>,
}

#[derive(Deserialize)]
struct TableUpdate {
    table: String,
    data: serde_json::Value,
}

// Schema IDs for each output topic, registered dynamically at startup
static SCHEMA_IDS: OnceLock<HashMap<String, i32>> = OnceLock::new();

fn main() {
    // Initialize logging
    env_logger::init();

    // Create Schema Registry client
    let mut client = SchemaRegistryClient::new();

    // Define schemas for each output topic
    let schemas = [
        ("orders", r#"{"type":"record","name":"Order","fields":[{"name":"order_id","type":"string"},{"name":"amount","type":"double"}]}"#),
        ("inventory", r#"{"type":"record","name":"Inventory","fields":[{"name":"product_id","type":"string"},{"name":"quantity","type":"int"}]}"#),
        ("customers", r#"{"type":"record","name":"Customer","fields":[{"name":"customer_id","type":"string"},{"name":"name","type":"string"}]}"#),
    ];

    let mut schema_ids = HashMap::new();

    // Register schemas and store their IDs
    for (topic, schema_str) in schemas {
        let subject = format!("{}-value", topic);
        let schema = Schema::new(schema_str.to_string(), SchemaFormat::Avro, vec![]);

        match client.create_schema(&subject, schema) {
            Ok(result) => {
                let id = result.id(); // SchemaId type
                schema_ids.insert(topic.to_string(), id.0); // Extract i32 from SchemaId wrapper
                info!("Registered schema for {} with ID {}", topic, id.0);
            }
            Err(e) => {
                error!("Failed to register schema for {}: {}", topic, e);
                panic!("Schema registration failed");
            }
        }
    }

    let _ = SCHEMA_IDS.set(schema_ids);
    info!("Starting fanout transform with schema IDs");

    on_record_written(route_updates);
}

fn write_update(
    update: &TableUpdate,
    schema_id: i32,
    writer: &mut RecordWriter,
    event: &WriteEvent,
) -> Result<(), Box<dyn Error>> {
    // Create Schema Registry wire format: [magic_byte, schema_id (4 bytes BE), data...]
    let mut value = vec![0u8; 5];
    value[0] = 0; // magic byte
    value[1..5].copy_from_slice(&schema_id.to_be_bytes());

    let data_bytes = serde_json::to_vec(&update.data)?;
    value.extend_from_slice(&data_bytes);

    let key = event.record.key().map(|k| k.to_vec());
    let record = BorrowedRecord::new(key.as_deref(), Some(&value));

    writer.write_with_options(record, WriteOptions::to_topic(&update.table))?;
    Ok(())
}

fn route_updates(event: WriteEvent, writer: &mut RecordWriter) -> Result<(), Box<dyn Error>> {
    let batch: BatchMessage = serde_json::from_slice(event.record.value().unwrap_or_default())?;
    let schema_ids = SCHEMA_IDS.get().unwrap();

    for update in batch.updates.iter() {
        if let Some(&schema_id) = schema_ids.get(&update.table) {
            write_update(update, schema_id, writer, &event)?;
        }
    }

    Ok(())
}