Add property to control shutdown timeout for processors

As a replacement for `KAFKA_STREAMS_DRAIN_TIMEOUT_DURATION`.

Signed-off-by: nscuro <[email protected]>

Author: nscuro

src/main/java/org/dependencytrack/event/kafka/processor/api/ProcessorManager.java

@@ -78,6 +78,8 @@
 import static org.dependencytrack.event.kafka.processor.api.ProcessorProperties.PROPERTY_RETRY_MULTIPLIER_DEFAULT;
 import static org.dependencytrack.event.kafka.processor.api.ProcessorProperties.PROPERTY_RETRY_RANDOMIZATION_FACTOR;
 import static org.dependencytrack.event.kafka.processor.api.ProcessorProperties.PROPERTY_RETRY_RANDOMIZATION_FACTOR_DEFAULT;
+import static org.dependencytrack.event.kafka.processor.api.ProcessorProperties.PROPERTY_SHUTDOWN_TIMEOUT_MS;
+import static org.dependencytrack.event.kafka.processor.api.ProcessorProperties.PROPERTY_SHUTDOWN_TIMEOUT_MS_DEFAULT;
 
 public class ProcessorManager implements AutoCloseable {
 
@@ -289,6 +291,11 @@ private ParallelStreamProcessor<byte[], byte[]> createParallelConsumer(final Str
             return Duration.ofMillis(delayMillis);
         });
 
+        final long shutdownTimeoutMs = Optional.ofNullable(properties.get(PROPERTY_SHUTDOWN_TIMEOUT_MS))
+                .map(Long::parseLong)
+                .orElse(PROPERTY_SHUTDOWN_TIMEOUT_MS_DEFAULT);
+        optionsBuilder.shutdownTimeout(Duration.ofMillis(shutdownTimeoutMs));
+
         if (Config.getInstance().getPropertyAsBoolean(Config.AlpineKey.METRICS_ENABLED)) {
             optionsBuilder
                     .meterRegistry(Metrics.getRegistry())

src/main/java/org/dependencytrack/event/kafka/processor/api/ProcessorProperties.java

@@ -36,6 +36,8 @@ final class ProcessorProperties {
     static final double PROPERTY_RETRY_RANDOMIZATION_FACTOR_DEFAULT = 0.3;
     static final String PROPERTY_RETRY_MAX_DELAY_MS = "retry.max.delay.ms";
     static final long PROPERTY_RETRY_MAX_DELAY_MS_DEFAULT = 60 * 1000; // 60s
+    static final String PROPERTY_SHUTDOWN_TIMEOUT_MS = "shutdown.timeout.ms";
+    static final long PROPERTY_SHUTDOWN_TIMEOUT_MS_DEFAULT = 10 * 1000; // 10s
 
     private ProcessorProperties() {
     }

src/main/resources/application.properties

@@ -453,6 +453,11 @@ application.id=dtrack-apiserver
 # alpine.kafka.processor.<name>.retry.randomization.factor=0.3
 # alpine.kafka.processor.<name>.retry.max.delay.ms=60000
 
+# Optional
+# Defines the timeout to wait for the processor to finish any pending work
+# prior to being shut down.
+# alpine.kafka.processor.<name>.shutdown.timeout.ms=10000
+
 # Optional
 # Allows for customization of the underlying Kafka consumer.
 # Refer to https://kafka.apache.org/documentation/#consumerconfigs for available options.