[olalekan-code]

Context

During a production MirrorMaker rebalance, a newly started worker was assigned task 1 after replaying the config topic. Its local ClusterConfigState contained the connector config but no task config for the assigned task. DistributedHerder passed the null map to Worker, which eventually failed with the opaque exception:

NullPointerException: Cannot invoke java.util.Map.size() because inputMap is null

The signature matches the failure family tracked in Apache KAFKA-10323 and KAFKA-13486.

Change

• resolve the task configuration once before dispatching to sink/source/exactly-once worker start paths
• reject a missing task configuration with an actionable ConnectException
• include the task ID, config offset, and whether the connector snapshot is marked inconsistent
• direct operators to refresh config state or reapply the connector configuration
• verify no Worker start method receives a null task configuration

Scope

This is a diagnostic and safety guard. It prevents the misleading TaskConfig(null) NPE and exposes the backing-store inconsistency directly. It does not automatically repair an incomplete compacted task-config generation. Correct automatic repair requires leader-side task reconfiguration/inter-worker coordination; DistributedHerderTest.testInconsistentConfigs already records that larger FIXME.

Validation

• focused incident regression test passed
• complete DistributedHerderTest class passed
• checkstyleMain, checkstyleTest, and spotbugsMain passed as dependencies of the Gradle test task
• git diff --check passed

Production image construction uses the Kafka 4.0.1 fork line, and the deployed fork commit contains origin/4.0 in its ancestry. A separate tw-kafka-connect pin should only follow after this change is reviewed and released.