KAFKA-16366: Refactor KTable source optimization

[chickenchickenlove]

Description

This PR refactors the DSL's optimization logic (reusing the input topic as the changelog) to leverage the public Topology#addReadOnlyStateStore API introduced in KIP-813, replacing the previous usage of internal builder methods.

Key Changes:

• Refactored TableSourceNode
  • Replaced the manual wiring of sources and stores (via addSource, addProcessor, connectSourceStoreAndTopic) with a call to topologyBuilder.addReadOnlyStateStore.
• Updated InternalTopologyBuilder:
  • Added an overloaded addReadOnlyStateStore method accepting a ProcessorSupplier with wildcard output types (?, ?).
  • Reasoning: The public PAPI addReadOnlyStateStore enforces Void, Void output types. However, the DSL's KTableSource needs to forward records downstream (emitting K, Change). The internal overload allows the DSL to reuse the "read-only" wiring logic while maintaining its forwarding behavior.
• Added Regression Tests:
  • Added TopologyTest#whenKTableSourceIsOptimizedThenTopologyShouldBeSerialPipeline - Verifies that the generated topology description remains identical to the pre-refactoring structure, ensuring strict backward compatibility.
  • Added TopologyTest#whenKTableSourceIsOptimizedThenItsStateStoreShouldNotLog - Confirms that the optimization is correctly applied (changelog logging is disabled).

Result

• Fixes https://issues.apache.org/jira/browse/KAFKA-16366

[chickenchickenlove]

Hi @mjsax !
Just a gentle ping on this PR.
When you have some time, could you please take a look? Thank you. 🙇‍♂️