Package ai.pipestream.common.interceptor

Class ApicurioConfigurator

java.lang.Object

ai.pipestream.common.interceptor.ApicurioConfigurator

All Implemented Interfaces:

io.smallrye.reactive.messaging.ClientCustomizer<Map<String,Object>>, jakarta.enterprise.inject.spi.Prioritized

@ApplicationScoped
@Identifier("smallrye-kafka")
public class ApicurioConfigurator
extends Object
implements io.smallrye.reactive.messaging.ClientCustomizer<Map<String,Object>>

Runtime configurator for Kafka clients in SmallRye Reactive Messaging.

This class implements the ClientCustomizer interface to customize Kafka producer and consumer configurations at runtime. It enforces consistent Protobuf serialization/deserialization settings across all Kafka channels and ensures proper integration with the Apicurio Registry.

Purpose

While the

invalid reference

ai.pipestream.kafka.connector.deployment.ProtobufKafkaProcessor

handles build-time configuration generation, this class provides runtime enforcement and fallback configuration. It acts as a safety net to ensure that all Kafka channels use the correct serializers and registry settings.

Configuration Applied

This configurator sets the following Kafka properties for all channels:

• Serializers/Deserializers: Forces Protobuf and UUID classes
• Apicurio Registry: URL, auto-registration, and schema lookup
• Return Class: Handled by build-time configuration (not here)

Thread Safety

This class is thread-safe as it only reads from injected configuration and modifies the provided configuration map. No mutable instance state is maintained between invocations.

Logging

Configuration changes are logged at DEBUG level, with sensitive values (passwords, secrets, tokens) automatically redacted in the logs.

Since:

0.2.10

See Also:

• ClientCustomizer
• invalid reference

  ai.pipestream.kafka.connector.deployment.ProtobufKafkaProcessor

Field Summary

Fields inherited from interface io.smallrye.reactive.messaging.ClientCustomizer

CLIENT_CONFIG_CUSTOMIZER_DEFAULT_PRIORITY

Constructor Summary

Constructors

Constructor	Description
ApicurioConfigurator()

Method Summary

Modifier and Type	Method	Description
Map<String,Object>	customize(String channelName, org.eclipse.microprofile.config.Config channelConfig, Map<String,Object> kafkaConfig)	Customizes Kafka client configuration for a specific channel.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface io.smallrye.reactive.messaging.ClientCustomizer

getPriority

Constructor Details

ApicurioConfigurator

public ApicurioConfigurator()

Method Details

customize

public Map<String,Object> customize(String channelName,
 org.eclipse.microprofile.config.Config channelConfig,
 Map<String,Object> kafkaConfig)

Customizes Kafka client configuration for a specific channel.

This method is called by SmallRye Reactive Messaging for each Kafka channel (both producers and consumers) to allow customization of the Kafka client properties. It enforces consistent Protobuf serialization and Apicurio Registry settings across all channels.

Configuration Strategy

The method applies a "sledgehammer" approach to ensure Protobuf serialization:

1. Force Serializers: Explicitly sets Protobuf and UUID classes
2. Apicurio Settings: Configures registry URL and auto-registration
3. Return Class: Handled by build-time configuration (not here)

Why This Approach

SmallRye Reactive Messaging may default to Jackson serialization if it cannot determine the correct serializer. This configurator prevents that by explicitly setting the Protobuf serializers, ensuring consistent message processing.

Specified by:

customize in interface io.smallrye.reactive.messaging.ClientCustomizer<Map<String,Object>>

Parameters:

channelName - The name of the messaging channel being configured

channelConfig - The MicroProfile Config for this specific channel

kafkaConfig - The mutable Kafka configuration map to customize

Returns:

The modified kafkaConfig map (same instance, modified in-place)

See Also:

• putAndLog(Map, String, Object)
• isSensitive(String)