Multi-Tenant Kafka Configuration

@Configuration
@Slf4j
public class TenantKafkaConfig {

    // Dynamic topic routing — single listener handles all tenants
    @KafkaListener(topicPattern = "tenant\..*\.order\.created",
                   groupId = "multi-tenant-processor")
    public void onTenantOrder(
        ConsumerRecord<String, OrderEvent> record,
        @Header(KafkaHeaders.RECEIVED_TOPIC) String topic
    ) {
        // Extract tenant from topic: tenant.{tenantId}.order.created
        String tenantId = topic.split("\.")[1];

        // Route to tenant-specific processing
        tenantOrderService.process(tenantId, record.value());
    }
}

// Publisher: routes to correct tenant topic
@Service
public class TenantOrderPublisher {

    public void publish(String tenantId, OrderEvent event) {
        String topic = String.format("tenant.%s.order.created", tenantId);
        kafkaTemplate.send(topic, event.getOrderId(), event);
    }
}

// Per-tenant consumer factory (for complete isolation)
@Bean
public Map<String, ConcurrentKafkaListenerContainerFactory<?, ?>> tenantFactories(
    Map<String, TenantConfig> tenants
) {
    return tenants.entrySet().stream()
        .collect(Collectors.toMap(
            Map.Entry::getKey,
            entry -> {
                var config = entry.getValue();
                Map<String, Object> props = new HashMap<>(commonProps);
                props.put(BOOTSTRAP_SERVERS_CONFIG, config.getKafkaBrokers());
                props.put(GROUP_ID_CONFIG, config.getTenantId() + "-processor");
                var cf = new DefaultKafkaConsumerFactory<>(props);
                var factory = new ConcurrentKafkaListenerContainerFactory<>();
                factory.setConsumerFactory(cf);
                return factory;
            }
        ));
}