[BUG] KafkaSink.doOutput can return while holding reentrantLock, and calls prepareTopicAndSchema() on every batch

Describe the bug

Two lifecycle problems in KafkaSink.doOutput, both found by inspection while
patching . Neither is fixed by that patch.

1. doOutput can return while holding the lock.

reentrantLock.lock();
if (records.isEmpty()) {
    return;                 // never unlocks
}
try {
    ...
} catch (Exception e) {
    LOG.error("Failed to setup the Kafka sink Plugin.", e);
    throw new RuntimeException(e.getMessage());
}
reentrantLock.unlock();

The empty-records early return never unlocks. ReentrantLock is reentrant per
thread, so a sink whose doOutput is always invoked from the same thread merely
accumulates hold count and appears to work. A sink invoked from a different
thread than the one that leaked the hold blocks forever. doOutput is called
from Pipeline.publishToSinks via sinkExecutorService, so the calling thread
is not guaranteed to be stable.

The unlock() is also not in a finally, so the exception path leaks the hold
as well — including the RuntimeException rethrown two lines above it.

2. prepareTopicAndSchema() runs on every doOutput. Upstream's own comment
says as much:

// TODO: Looks like this call to prepareTopicAndSchema is unnecessary as it is
// done in createProducer().
prepareTopicAndSchema();

With create_topic: true this constructs a TopicService and an admin client
per batch, then closes them — the same create-per-call shape as , and
per-batch admin client churn against the broker.

To Reproduce

Steps to reproduce the behavior:

For the lock (code inspection; the concrete failure needs doOutput to be
dispatched across threads):

1. Read KafkaSink.doOutput — the records.isEmpty() branch returns between
   lock() and unlock().
2. Configure a kafka sink on a pipeline whose buffer read can return an empty
   collection, so the early return is taken.
3. Have doOutput subsequently invoked from a different thread of
   sinkExecutorService than the one that took the early return.
4. That call blocks indefinitely on reentrantLock.lock().

For the admin client churn:

1. Configure a kafka sink with topic.create_topic: true.
2. Send sustained traffic.
3. Observe a TopicService and Kafka AdminClient constructed and closed once
   per batch rather than once per sink.

Expected behavior

doOutput should release reentrantLock on every exit path — unlock() in a
finally, or lock() after the empty check. prepareTopicAndSchema() should
run once during initialisation, as the existing TODO suggests.

Screenshots

N/A.

Environment (please complete the following information):

• OS: Ubuntu 24.04
• Version: OpenSearch Data Prepper 2.16.0 (docker.io/opensearchproject/data-prepper:2.16.0). Also verified present on main at commit 7222e0d0f.
• Kafka: docker.io/apache/kafka:4.3.1, KRaft, PLAINTEXT listener

Additional context

We have not observed the lock deadlock in production — our sinks appear to be
called from a stable thread, which masks it. Reporting it because the current
code is correct only by accident of the caller's threading, and #7131 moves
producer setup out of doOutput, which is a natural moment to also move
prepareTopicAndSchema() and tidy the lock.

Read KafkaSink.doOutput first, then trace Pipeline.publishToSinks and sinkExecutorService to understand lifecycle and thread ownership. Inspect the initialization path around createProducer() and prepareTopicAndSchema(). Done means every doOutput exit path releases the lock and topic/schema preparation is not repeated for each batch.