Kafka Consumer Lifecycle & Health Checks

@Configuration
@Slf4j
public class KafkaHealthConfig implements ApplicationListener<ListenerContainerIdleEvent> {

    // Called when a listener container is idle (no messages for idle.interval.ms)
    @Override
    public void onApplicationEvent(ListenerContainerIdleEvent event) {
        log.info("Consumer idle: listenerId={}, idleFor={}ms",
            event.getListenerId(), event.getIdleTime());
    }

    @Bean
    public KafkaListenerEndpointRegistry kafkaRegistry() {
        return new KafkaListenerEndpointRegistry();
    }
}

// Programmatic container control
@RestController
@RequestMapping("/admin/kafka")
public class KafkaAdminController {

    @Autowired
    private KafkaListenerEndpointRegistry registry;

    @PostMapping("/pause/{listenerId}")
    public void pauseListener(@PathVariable String listenerId) {
        registry.getListenerContainer(listenerId).pause();
    }

    @PostMapping("/resume/{listenerId}")
    public void resumeListener(@PathVariable String listenerId) {
        registry.getListenerContainer(listenerId).resume();
    }

    @GetMapping("/status")
    public Map<String, String> status() {
        return registry.getListenerContainerIds().stream()
            .collect(Collectors.toMap(
                id -> id,
                id -> registry.getListenerContainer(id).isRunning() ? "RUNNING" : "STOPPED"
            ));
    }
}