Migrate VulnerabilityScanResultProcessor from Kafka Streams to Parallel Consumer

pom.xml

@@ -288,16 +288,7 @@
             <artifactId>kafka-clients</artifactId>
             <version>${lib.kafka.version}</version>
         </dependency>
-        <dependency>
-            <groupId>org.apache.kafka</groupId>
-            <artifactId>kafka-streams</artifactId>
-            <version>${lib.kafka.version}</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.kafka</groupId>
-            <artifactId>kafka-streams-test-utils</artifactId>
-            <version>${lib.kafka.version}</version>
-        </dependency>
+
         <!-- shedlock dependencies -->
         <dependency>
             <groupId>net.javacrumbs.shedlock</groupId>

src/main/java/org/dependencytrack/common/ConfigKey.java

@@ -40,17 +40,8 @@ public enum ConfigKey implements Config.Key {
     KEY_STORE_PATH("kafka.keystore.path", ""),
 
     KEY_STORE_PASSWORD("kafka.keystore.password", ""),
-    KAFKA_NUM_STREAM_THREADS("kafka.num.stream.threads", 1),
     KAFKA_TOPIC_PREFIX("kafka.topic.prefix", ""),
     KAFKA_PRODUCER_DRAIN_TIMEOUT_DURATION("kafka.producer.drain.timeout.duration", "PT30S"),
-    KAFKA_STREAMS_DESERIALIZATION_EXCEPTION_THRESHOLD_COUNT("kafka.streams.deserialization.exception.threshold.count", "5"),
-    KAFKA_STREAMS_DESERIALIZATION_EXCEPTION_THRESHOLD_INTERVAL("kafka.streams.deserialization.exception.threshold.interval", "PT30M"),
-    KAFKA_STREAMS_METRICS_RECORDING_LEVEL("kafka.streams.metrics.recording.level", "INFO"),
-    KAFKA_STREAMS_PRODUCTION_EXCEPTION_THRESHOLD_COUNT("kafka.streams.production.exception.threshold.count", "5"),
-    KAFKA_STREAMS_PRODUCTION_EXCEPTION_THRESHOLD_INTERVAL("kafka.streams.production.exception.threshold.interval", "PT30H"),
-    KAFKA_STREAMS_TRANSIENT_PROCESSING_EXCEPTION_THRESHOLD_COUNT("kafka.streams.transient.processing.exception.threshold.count", "50"),
-    KAFKA_STREAMS_TRANSIENT_PROCESSING_EXCEPTION_THRESHOLD_INTERVAL("kafka.streams.transient.processing.exception.threshold.interval", "PT30M"),
-    KAFKA_STREAMS_DRAIN_TIMEOUT_DURATION("kafka.streams.drain.timeout.duration", "PT30S"),
 
     CRON_EXPRESSION_FOR_PORTFOLIO_METRICS_TASK("task.cron.metrics.portfolio", "10 * * * *"),
     CRON_EXPRESSION_FOR_VULNERABILITY_METRICS_TASK("task.cron.metrics.vulnerability", "40 * * * *"),

src/main/java/org/dependencytrack/common/MdcKeys.java

@@ -23,10 +23,12 @@
  */
 public final class MdcKeys {
 
+    public static final String MDC_COMPONENT_UUID = "componentUuid";
     public static final String MDC_KAFKA_RECORD_TOPIC = "kafkaRecordTopic";
     public static final String MDC_KAFKA_RECORD_PARTITION = "kafkaRecordPartition";
     public static final String MDC_KAFKA_RECORD_OFFSET = "kafkaRecordOffset";
     public static final String MDC_KAFKA_RECORD_KEY = "kafkaRecordKey";
+    public static final String MDC_SCAN_TOKEN = "scanToken";
 
     private MdcKeys() {
     }

src/main/java/org/dependencytrack/event/kafka/KafkaEventConverter.java

@@ -58,7 +58,7 @@
  * Utility class to convert {@link alpine.event.framework.Event}s and {@link alpine.notification.Notification}s
  * to {@link KafkaEvent}s.
  */
-final class KafkaEventConverter {
+public final class KafkaEventConverter {
 
     private KafkaEventConverter() {
     }
@@ -80,7 +80,7 @@ private KafkaEventConverter() {
         return convert(protoNotification);
     }
 
-    static KafkaEvent<?, ?> convert(final Notification notification) {
+    public static KafkaEvent<?, ?> convert(final Notification notification) {
         final Topic<String, Notification> topic = extractDestinationTopic(notification);
 
         final String recordKey;

src/main/java/org/dependencytrack/event/kafka/KafkaProducerInitializer.java

@@ -107,6 +107,7 @@ private static Producer<byte[], byte[]> createProducer() {
         properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
         properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, CompressionType.SNAPPY.name);
         properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
+        properties.put(ProducerConfig.LINGER_MS_CONFIG, "100");
         properties.put(ProducerConfig.ACKS_CONFIG, "all");
         if (Config.getInstance().getPropertyAsBoolean(ConfigKey.KAFKA_TLS_ENABLED)) {
             properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, Config.getInstance().getProperty(ConfigKey.KAFKA_TLS_PROTOCOL));

src/main/java/org/dependencytrack/event/kafka/KafkaTopics.java

@@ -53,6 +53,7 @@ public final class KafkaTopics {
     public static final Topic<String, AnalysisResult> REPO_META_ANALYSIS_RESULT;
     public static final Topic<ScanKey, ScanCommand> VULN_ANALYSIS_COMMAND;
     public static final Topic<ScanKey, ScanResult> VULN_ANALYSIS_RESULT;
+    public static final Topic<String, ScanResult> VULN_ANALYSIS_RESULT_PROCESSED;
 
     public static final Topic<String, Notification> NOTIFICATION_PROJECT_VULN_ANALYSIS_COMPLETE;
     public static final Topic<String, EpssItem> NEW_EPSS;
@@ -78,6 +79,7 @@ public final class KafkaTopics {
         REPO_META_ANALYSIS_RESULT = new Topic<>("dtrack.repo-meta-analysis.result", Serdes.String(), new KafkaProtobufSerde<>(AnalysisResult.parser()));
         VULN_ANALYSIS_COMMAND = new Topic<>("dtrack.vuln-analysis.component", new KafkaProtobufSerde<>(ScanKey.parser()), new KafkaProtobufSerde<>(ScanCommand.parser()));
         VULN_ANALYSIS_RESULT = new Topic<>("dtrack.vuln-analysis.result", new KafkaProtobufSerde<>(ScanKey.parser()), new KafkaProtobufSerde<>(ScanResult.parser()));
+        VULN_ANALYSIS_RESULT_PROCESSED = new Topic<>("dtrack.vuln-analysis.result.processed", Serdes.String(), new KafkaProtobufSerde<>(ScanResult.parser()));
         NOTIFICATION_PROJECT_VULN_ANALYSIS_COMPLETE = new Topic<>("dtrack.notification.project-vuln-analysis-complete", Serdes.String(), NOTIFICATION_SERDE);
         NEW_EPSS = new Topic<>("dtrack.epss", Serdes.String(), new KafkaProtobufSerde<>(EpssItem.parser()));
     }