{"record":{"id":"ba5363202a06335d","repo":"apache/druid","slug":"already-shut-down-not-starting-again","errorCode":null,"errorMessage":"Already shut down, not starting again","messagePattern":"Already shut down, not starting again","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"extensions-core/kafka-extraction-namespace/src/main/java/org/apache/druid/query/lookup/KafkaLookupExtractorFactory.java","lineNumber":161,"sourceCode":"  {\n    return connectTimeout;\n  }\n\n  public boolean isInjective()\n  {\n    return injective;\n  }\n\n  @Override\n  public boolean start()\n  {\n    synchronized (started) {\n      if (started.get()) {\n        LOG.warn(\"Already started, not starting again\");\n        return true;\n      }\n      if (executorService.isShutdown()) {\n        LOG.warn(\"Already shut down, not starting again\");\n        return false;\n      }\n      verifyKafkaProperties();\n\n      final String topic = getKafkaTopic();\n      LOG.debug(\"About to listen to topic [%s] with group.id [%s]\", topic, factoryId);\n      // this creates a ConcurrentMap\n      cacheHandler = cacheManager.createCache();\n      final Map<String, String> map = cacheHandler.getCache();\n      mapRef.set(map);\n\n\n      final CountDownLatch startingReads = new CountDownLatch(1);\n\n      final ListenableFuture<?> future = executorService.submit(() -> {\n        final Consumer<String, String> consumer = getConsumer();\n        consumer.subscribe(Collections.singletonList(topic));\n        try {","sourceCodeStart":143,"sourceCodeEnd":179,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/extensions-core/kafka-extraction-namespace/src/main/java/org/apache/druid/query/lookup/KafkaLookupExtractorFactory.java#L143-L179","documentation":"KafkaLookupExtractorFactory.start() returns false instead of starting the Kafka consumer when the executor service backing the lookup has already been shut down. The factory keeps an internal 'started' flag and executor; once stop()/shutdown has run, a restart is deliberately refused so lookups never half-start against dead resources. It is a returned boolean (logged at warn), not an exception.","triggerScenarios":"Calling start() after stop() was called on the same KafkaLookupExtractorFactory instance; the executorService is in shutdown state, so start() returns false.","commonSituations":"Lookup manager re-initializing a cached lookup instance that was previously torn down; restarting a suspended task after the lookup factory was closed; mismanaged lifecycle where the same factory object is reused across start/stop cycles.","solutions":["Create a new KafkaLookupExtractorFactory instance instead of reusing a stopped one.","Check the return value of start() and rebuild the factory if it returns false.","Do not call stop() on the factory unless the lookup is permanently retired.","Verify lifecycle ordering in the lookup extraction manager so start() is never invoked after shutdown."],"exampleFix":"// before\nfactory.stop();\nboolean ok = factory.start(); // returns false: \"Already shut down\"\n// after\nfactory.stop();\nfactory = new KafkaLookupExtractorFactory(bootstrapServers, topic, kafkaProperties);\nboolean ok = factory.start();","handlingStrategy":"validation","validationCode":"if (factory.isStarted()) return; // or rebuild factory before calling start()\nboolean ok = factory.start();\nif (!ok) { factory = new KafkaLookupExtractorFactory(bootstrap, topic, props); ok = factory.start(); }","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Treat factory instances as single-use: new instance per lifecycle","Check the boolean return of start() rather than assuming success","Never call stop() for transient suspensions"],"tags":["kafka","lookup","lifecycle","shutdown"],"backgroundTag":"invalid-state-transition","analyzedSha":"9b90983fd291f26935af934383ce360473179e4d","analyzedAt":"2026-09-07T13:32:30.957Z","contentChangedAt":"2026-09-07T13:32:30.957Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}