Kafka Consumer Lifecycle & Health Checks

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

    // Published when a container has received no records for the configured idleEventInterval
    @Override
    public void onApplicationEvent(ListenerContainerIdleEvent event) {
        log.info("Consumer idle: listenerId={}, idleFor={}ms",
            event.getListenerId(), event.getIdleTime());
    }
}

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

    // The registry is an infrastructure bean that the framework declares automatically
    @Autowired
    private KafkaListenerEndpointRegistry registry;

    @PostMapping("/pause/{listenerId}")
    public void pauseListener(@PathVariable String listenerId) {
        registry.getListenerContainer(listenerId).pause();   // takes effect before the next poll()
    }

    @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"
            ));
    }
}