[BUG] Kafka sink throws NullPointerException on every shutdown when thread_wait_time is not set

Describe the bug

KafkaSinkConfig.threadWaitTime is a Long with no default value, and
KafkaSink.shutdown() unboxes it into
ExecutorService.awaitTermination(long, TimeUnit):

@JsonProperty("thread_wait_time")
private Long threadWaitTime;                    // KafkaSinkConfig.java

private long calculateLongestThreadWaitingTime() {
    return kafkaSinkConfig.getThreadWaitTime(); // KafkaSink.java — NPE when unset
}

thread_wait_time is not marked @NotNull, is not documented as required, and
has no default, so any pipeline that omits it throws NullPointerException on
every shutdown. The exception surfaces on a sink worker thread rather than
the pipeline thread, so the pipeline never shuts down cleanly and ingestion
stops with nothing in the log naming a cause.

To Reproduce

Steps to reproduce the behavior:

1. Configure any pipeline with a kafka sink and omit thread_wait_time:

   sink:
     - kafka:
         bootstrap_servers: ["kafka-server:9092"]
         encryption:
           type: none
         serde_format: json
         topic:
           name: my-topic
   

2. Start Data Prepper and let the pipeline process at least one batch.
3. Stop Data Prepper (SIGTERM, or stop the container).
4. See the error in the log:

   Exception in thread "linux-syslog-messages-sink-worker-26-thread-1"
   java.lang.NullPointerException: Cannot invoke "java.lang.Long.longValue()"
     because the return value of KafkaSinkConfig.getThreadWaitTime() is null
       at KafkaSink.calculateLongestThreadWaitingTime(KafkaSink.java:194)
       at KafkaSink.shutdown(KafkaSink.java:178)
   

Expected behavior

thread_wait_time is optional, so omitting it should apply a documented default
and shut the sink down cleanly. Adding thread_wait_time: 5000 confirms this is
the only thing missing — the NPE is replaced by
KafkaSink - Producer shutdown successfully....

Screenshots

N/A — log output inline above.

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), running under podman. Also verified present on main at commit 7222e0d0f (2.17.0-SNAPSHOT).
• Kafka: docker.io/apache/kafka:4.3.1, KRaft, PLAINTEXT listener

Additional context

Two possible fixes, and we have applied both locally:

1. A default on the config field —
   private Long threadWaitTime = DEFAULT_THREAD_WAIT_TIME_MS; with
   static final long DEFAULT_THREAD_WAIT_TIME_MS = 1000L. 1000 ms is the value
   already used throughout the existing sink tests and
   sample-pipelines-sink.yaml.
2. A null guard at the unboxing site, so an explicit thread_wait_time: null
   cannot reintroduce the crash.

Happy to open a PR with both plus unit tests. Note that a restart timer is not a
usable workaround: with this NPE present, shutting down all pipelines took over
11 minutes in our deployment (Workers did not terminate in PT30S, forcing termination, repeated per pipeline).

Start with KafkaSinkConfig.java and KafkaSink.java, especially getThreadWaitTime(), calculateLongestThreadWaitingTime(), and shutdown(). Review the existing sink tests and sample-pipelines-sink.yaml for the established 1000 ms value. Done means an omitted or explicitly null thread_wait_time no longer causes shutdown to throw, with tests covering the configuration and clean shutdown behavior.