Welcome to kpipe Discussions!

[eschizoid]

Type-Safe Registry Modernization Proposal

Objective

Improve compile-time safety and developer experience when registering and using message processors and sinks.

Ideas for Improving Type Safety in KPipe Registries

The current implementation of MessageProcessorRegistry and MessageSinkRegistry relies heavily on String
identifiers and byte[] or Object (via Map<String, Object> or GenericRecord) for message processing. While this
provides flexibility, it sacrifices compile-time type safety.

1. Typed Registry Keys (Type-Safe Identifiers)

Instead of using plain String names, introduce a typed RegistryKey<T> class. This ensures that when you retrieve a
processor or sink, you know exactly what data type it operates on.

public record RegistryKey<T>(String name, Class<T> type) {
}

// Usage in MessageProcessorRegistry
public <T> void registerOperator(RegistryKey<T> key, UnaryOperator<T> operator) {
  // Store in a map keyed by RegistryKey or name
}

public <T> UnaryOperator<T> getOperator(RegistryKey<T> key) {
  // Returns typed operator, avoiding manual casting
}

// Pre-defined keys
public static final RegistryKey<Map<String, Object>> JSON_TIMESTAMP =
        new RegistryKey<>("addTimestamp", Map.class);

2. Fluent "Builder" for Pipelines

Create a fluent API that guides the user through the pipeline construction based on types. This prevents mixing
incompatible processors (e.g., trying to use an Avro operator in a JSON pipeline).

// Example of a type-safe pipeline builder
var pipeline = registry.jsonPipelineBuilder()
                .add(JsonMessageProcessor.addTimestampOperator("ts"))
                .add(myCustomJsonOperator)
                .build(); // Returns Function<byte[], byte[]>

3. Strongly Typed Sink Registry

The MessageSinkRegistry can be made safer by requiring the key and value types at registration time.

public class MessageSinkRegistry {
  // Use a multi-key or nested map to store by (Name, KeyClass, ValueClass)
  public <K, V> void register(String name, Class<K> keyType, Class<V> valueType, MessageSink<K, V> sink) { ...}

  public <K, V> MessageSink<K, V> get(String name, Class<K> keyType, Class<V> valueType) {
    // Verification happens here instead of at the call site
  }
}

4. Sealed Interface for Message Formats

Since MessageFormat is currently an enum, it's hard to attach specific type information to each format. Moving to a
sealed class/interface hierarchy would allow better type bounds.

public sealed interface MessageFormat<T> permits JsonFormat, AvroFormat {
  T deserialize(byte[] bytes);

  byte[] serialize(T data);
}

public final class JsonFormat implements MessageFormat<Map<String, Object>> { ...
}

public final class AvroFormat implements MessageFormat<GenericRecord> { ...
}

5. Annotation Processing or Record-Based Mapping

For users who prefer POJOs over Map or GenericRecord, provide a way to register processors based on Java records or
classes.

// Logic that automatically handles SerDe based on the class
registry.registerProcessor("userMasker",UserRecord .class, user ->
        new

UserRecord(user.id(), "MASKED",user.

email())
        );

6. Enum-Based Registry (Static Type Safety)

If the set of processors is known at compile-time, using an Enum that implements an interface is the ultimate way to
ensure type safety.

public enum MyProcessors implements UnaryOperator<Map<String, Object>> {
  TIMESTAMP(JsonMessageProcessor.addTimestampOperator("ts")),
  SOURCE(JsonMessageProcessor.addFieldOperator("src", "app"));

  private final UnaryOperator<Map<String, Object>> op;

  MyProcessors(UnaryOperator<Map<String, Object>> op) {
    this.op = op;
  }

  @Override
  public Map<String, Object> apply(Map<String, Object> t) {
    return op.apply(t);
  }
}

// Registry then accepts the Enum type
public void registerPipeline(MyProcessors... steps) { ...}

Roadmap

Phase 1: Foundation (Type-Safe Identifiers)

• Introduce RegistryKey<T> record to replace String names in MessageProcessorRegistry and MessageSinkRegistry.
• Update register and get methods to accept RegistryKey<T>.
• Deprecate raw String registration methods.

Phase 2: Pipeline Safety

• Implement a fluent PipelineBuilder that uses RegistryKey<T> to ensure only compatible operators are chained.
• Transition MessageFormat from an enum to a sealed interface hierarchy.

Phase 3: Sink & POJO Support

• Modernize MessageSinkRegistry to require key and value type verification at registration time.
• Add support for record-based mapping to allow users to work with POJOs instead of Map or GenericRecord.

Success Metrics

• Zero ClassCastException at runtime when accessing registered components.
• Compiler errors when mixing incompatible processor types in the same pipeline.
• Reduced boilerplate for end-users when defining complex pipelines.