Merge pull request #553 from DependencyTrack/issue-907-2

nscuro · web-flow · commit 8ff235b9a68b · 2024-03-22T15:29:15.000+01:00
Migrate `MirrorVulnerabilityProcessor` from Kafka Streams to Parallel Consumer
diff --git a/src/main/java/org/dependencytrack/event/kafka/processor/ProcessorInitializer.java b/src/main/java/org/dependencytrack/event/kafka/processor/ProcessorInitializer.java
@@ -1,6 +1,7 @@
 package org.dependencytrack.event.kafka.processor;
 
 import alpine.common.logging.Logger;
+import org.dependencytrack.event.kafka.KafkaTopics;
 import org.dependencytrack.event.kafka.processor.api.ProcessorManager;
 
 import javax.servlet.ServletContextEvent;
@@ -16,7 +17,8 @@ public class ProcessorInitializer implements ServletContextListener {
     public void contextInitialized(final ServletContextEvent event) {
         LOGGER.info("Initializing processors");
 
-        // TODO: Register processor here!
+        PROCESSOR_MANAGER.registerProcessor(VulnerabilityMirrorProcessor.PROCESSOR_NAME,
+                KafkaTopics.NEW_VULNERABILITY, new VulnerabilityMirrorProcessor());
 
         PROCESSOR_MANAGER.startAll();
     }
diff --git a/src/main/java/org/dependencytrack/event/kafka/processor/VulnerabilityMirrorProcessor.java b/src/main/java/org/dependencytrack/event/kafka/processor/VulnerabilityMirrorProcessor.java
@@ -1,16 +1,14 @@
-package org.dependencytrack.event.kafka.streams.processor;
+package org.dependencytrack.event.kafka.processor;
 
 import alpine.common.logging.Logger;
-import alpine.common.metrics.Metrics;
 import com.github.packageurl.MalformedPackageURLException;
 import com.github.packageurl.PackageURL;
-import io.micrometer.core.instrument.Timer;
 import org.apache.commons.lang3.StringUtils;
-import org.apache.kafka.streams.processor.api.Processor;
-import org.apache.kafka.streams.processor.api.Record;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.cyclonedx.proto.v1_4.Bom;
 import org.cyclonedx.proto.v1_4.Component;
 import org.cyclonedx.proto.v1_4.VulnerabilityAffects;
+import org.dependencytrack.event.kafka.processor.api.Processor;
 import org.dependencytrack.model.Vulnerability;
 import org.dependencytrack.model.VulnerableSoftware;
 import org.dependencytrack.parser.dependencytrack.ModelConverterCdxToVuln;
@@ -27,19 +25,18 @@
 import java.util.List;
 import java.util.Optional;
 
+/**
+ * A {@link Processor} that ingests vulnerability data from CycloneDX Bill of Vulnerabilities.
+ */
+public class VulnerabilityMirrorProcessor implements Processor<String, Bom> {
 
-public class MirrorVulnerabilityProcessor implements Processor<String, Bom, Void, Void> {
+    static final String PROCESSOR_NAME = "vuln.mirror";
 
-    private static final Logger LOGGER = Logger.getLogger(MirrorVulnerabilityProcessor.class);
-    private static final Timer TIMER = Timer.builder("vuln_mirror_processing")
-            .description("Time taken to process mirrored vulnerabilities")
-            .register(Metrics.getRegistry());
+    private static final Logger LOGGER = Logger.getLogger(VulnerabilityMirrorProcessor.class);
 
     @Override
-    public void process(final Record<String, Bom> record) {
-        final Timer.Sample timerSample = Timer.start();
-
-        try (QueryManager qm = new QueryManager().withL2CacheDisabled()) {
+    public void process(final ConsumerRecord<String, Bom> record) {
+        try (QueryManager qm = new QueryManager()) {
             LOGGER.debug("Synchronizing Mirrored Vulnerability : " + record.key());
             Bom bom = record.value();
             String key = record.key();
@@ -112,11 +109,6 @@ public void process(final Record<String, Bom> record) {
                 synchronizedVulnerability.setVulnerableSoftware(reconciledVsList);
             }
             qm.persist(synchronizedVulnerability);
-        } catch (Exception e) {
-            // TODO: Send record to a dead letter topic.
-            LOGGER.error("Synchronizing vulnerability %s failed".formatted(record.key()), e);
-        } finally {
-            timerSample.stop(TIMER);
         }
     }
 
@@ -230,8 +222,8 @@ public VulnerableSoftware mapAffectedRangeToVulnerableSoftware(final QueryManage
 
         for (final Constraint constraint : vers.constraints()) {
             if (constraint.version() == null
-                    || constraint.version().equals("0")
-                    || constraint.version().equals("*")) {
+                || constraint.version().equals("0")
+                || constraint.version().equals("*")) {
                 // Semantically, ">=0" is equivalent to versionStartIncluding=null,
                 // and ">0" is equivalent to versionStartExcluding=null.
                 //
@@ -253,12 +245,12 @@ public VulnerableSoftware mapAffectedRangeToVulnerableSoftware(final QueryManage
         }
 
         if (versionStartIncluding == null && versionStartExcluding == null
-                && versionEndIncluding == null && versionEndExcluding == null) {
+            && versionEndIncluding == null && versionEndExcluding == null) {
             LOGGER.warn("Unable to assemble a version range from %s for %s".formatted(vers, vulnId));
             return null;
         }
         if ((versionStartIncluding != null || versionStartExcluding != null)
-                && (versionEndIncluding == null && versionEndExcluding == null)) {
+            && (versionEndIncluding == null && versionEndExcluding == null)) {
             LOGGER.warn("Skipping indefinite version range assembled from %s for %s".formatted(vers, vulnId));
             return null;
         }
diff --git a/src/main/java/org/dependencytrack/event/kafka/streams/KafkaStreamsTopologyFactory.java b/src/main/java/org/dependencytrack/event/kafka/streams/KafkaStreamsTopologyFactory.java
@@ -23,7 +23,6 @@
 import org.dependencytrack.event.ProjectPolicyEvaluationEvent;
 import org.dependencytrack.event.kafka.KafkaTopics;
 import org.dependencytrack.event.kafka.streams.processor.DelayedBomProcessedNotificationProcessor;
-import org.dependencytrack.event.kafka.streams.processor.MirrorVulnerabilityProcessor;
 import org.dependencytrack.event.kafka.streams.processor.RepositoryMetaResultProcessor;
 import org.dependencytrack.event.kafka.streams.processor.VulnerabilityScanResultProcessor;
 import org.dependencytrack.model.VulnerabilityScan;
@@ -224,12 +223,6 @@ Topology createTopology() {
                                 .withName("consume_from_%s_topic".formatted(KafkaTopics.REPO_META_ANALYSIS_RESULT.name())))
                 .process(RepositoryMetaResultProcessor::new, Named.as("process_repo_meta_analysis_result"));
 
-        streamsBuilder
-                .stream(KafkaTopics.NEW_VULNERABILITY.name(),
-                        Consumed.with(KafkaTopics.NEW_VULNERABILITY.keySerde(), KafkaTopics.NEW_VULNERABILITY.valueSerde())
-                                .withName("consume_from_%s_topic".formatted(KafkaTopics.NEW_VULNERABILITY.name())))
-                .process(MirrorVulnerabilityProcessor::new, Named.as("process_mirror_vulnerability"));
-
         return streamsBuilder.build(streamsProperties);
     }
 
diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties
@@ -491,6 +491,19 @@ kafka.streams.transient.processing.exception.threshold.interval=PT30M
 # Refer to https://kafka.apache.org/documentation/#consumerconfigs for available options.
 # alpine.kafka.processor.<name>.consumer.<consumer.config.name>=
 
+# Required
+# Configures the Kafka processor responsible for ingesting mirrored vulnerability
+# data from the dtrack.vulnerability topic. The processor only occasionally receives
+# records, such that high concurrency is usually not justified.
+alpine.kafka.processor.vuln.mirror.max.concurrency=-1
+alpine.kafka.processor.vuln.mirror.processing.order=partition
+alpine.kafka.processor.vuln.mirror.retry.initial.delay.ms=3000
+alpine.kafka.processor.vuln.mirror.retry.multiplier=2
+alpine.kafka.processor.vuln.mirror.retry.randomization.factor=0.3
+alpine.kafka.processor.vuln.mirror.retry.max.delay.ms=180000
+alpine.kafka.processor.vuln.mirror.consumer.group.id=dtrack-apiserver-processor
+alpine.kafka.processor.vuln.mirror.consumer.auto.offset.reset=earliest
+
 # Scheduling tasks after 3 minutes (3*60*1000) of starting application
 task.scheduler.initial.delay=180000
 
diff --git a/src/test/java/org/dependencytrack/event/kafka/processor/VulnerabilityMirrorProcessorTest.java b/src/test/java/org/dependencytrack/event/kafka/processor/VulnerabilityMirrorProcessorTest.java
@@ -1,55 +1,26 @@
-package org.dependencytrack.event.kafka.streams.processor;
+package org.dependencytrack.event.kafka.processor;
 
-import org.apache.kafka.common.serialization.Serdes;
-import org.apache.kafka.common.serialization.StringSerializer;
-import org.apache.kafka.streams.StreamsBuilder;
-import org.apache.kafka.streams.TestInputTopic;
-import org.apache.kafka.streams.TopologyTestDriver;
-import org.apache.kafka.streams.kstream.Consumed;
-import org.cyclonedx.proto.v1_4.Bom;
-import org.dependencytrack.PersistenceCapableTest;
-import org.dependencytrack.event.kafka.serialization.KafkaProtobufSerde;
-import org.dependencytrack.event.kafka.serialization.KafkaProtobufSerializer;
 import org.dependencytrack.model.Severity;
 import org.dependencytrack.model.Vulnerability;
 import org.dependencytrack.persistence.CweImporter;
-import org.dependencytrack.util.KafkaTestUtil;
-import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.dependencytrack.util.KafkaTestUtil.generateBomFromJson;
 
-public class MirrorVulnerabilityProcessorTest extends PersistenceCapableTest {
-
-    private TopologyTestDriver testDriver;
-    private TestInputTopic<String, Bom> inputTopic;
+public class VulnerabilityMirrorProcessorTest extends AbstractProcessorTest {
 
     @Before
-    public void setUp() throws Exception {
-        final var streamsBuilder = new StreamsBuilder();
-        streamsBuilder
-                .stream("input-topic", Consumed
-                        .with(Serdes.String(), new KafkaProtobufSerde<>(Bom.parser())))
-                .process(MirrorVulnerabilityProcessor::new);
-
-        testDriver = new TopologyTestDriver(streamsBuilder.build());
-        inputTopic = testDriver.createInputTopic("input-topic",
-                new StringSerializer(), new KafkaProtobufSerializer<>());
+    public void before() throws Exception {
+        super.before();
 
         new CweImporter().processCweDefinitions(); // Required for CWE mapping
     }
 
-    @After
-    public void tearDown() {
-        if (testDriver != null) {
-            testDriver.close();
-        }
-    }
-
     @Test
     public void testProcessNvdVuln() throws Exception {
-        inputTopic.pipeInput("NVD/CVE-2022-40489", KafkaTestUtil.generateBomFromJson("""
+        final var bovJson = """
                 {
                   "components": [
                     {
@@ -91,7 +62,10 @@ public void testProcessNvdVuln() throws Exception {
                     { "url": "https://github.com/thinkcmf/thinkcmf/issues/736" }
                   ]
                 }
-                """));
+                """;
+
+        final var processor = new VulnerabilityMirrorProcessor();
+        processor.process(aConsumerRecord("NVD/CVE-2022-40489", generateBomFromJson(bovJson)).build());
 
         final Vulnerability vuln = qm.getVulnerabilityByVulnId("NVD", "CVE-2022-40489");
         assertThat(vuln).isNotNull();
@@ -160,7 +134,7 @@ public void testProcessNvdVuln() throws Exception {
 
     @Test
     public void testProcessGitHubVuln() throws Exception {
-        inputTopic.pipeInput("GITHUB/GHSA-fxwm-579q-49qq", KafkaTestUtil.generateBomFromJson("""
+        final var bovJson = """
                 {
                   "components": [
                     {
@@ -223,7 +197,10 @@ public void testProcessGitHubVuln() throws Exception {
                     { "url": "https://github.com/advisories/GHSA-fxwm-579q-49qq" }
                   ]
                 }
-                """));
+                """;
+
+        final var processor = new VulnerabilityMirrorProcessor();
+        processor.process(aConsumerRecord("GITHUB/GHSA-fxwm-579q-49qq", generateBomFromJson(bovJson)).build());
 
         final Vulnerability vuln = qm.getVulnerabilityByVulnId("GITHUB", "GHSA-fxwm-579q-49qq");
         assertThat(vuln).isNotNull();
@@ -375,7 +352,7 @@ public void testProcessGitHubVuln() throws Exception {
 
     @Test
     public void testProcessOsvVuln() throws Exception {
-        inputTopic.pipeInput("OSV/GHSA-2cc5-23r7-vc4v", KafkaTestUtil.generateBomFromJson("""
+        final var bovJson = """
                 {
                   "components": [
                     {
@@ -427,7 +404,10 @@ public void testProcessOsvVuln() throws Exception {
                     { "url": "https://github.com/ratpack/ratpack/blob/29434f7ac6fd4b36a4495429b70f4c8163100332/ratpack-session/src/main/java/ratpack/session/clientside/ClientSideSessionConfig.java#L29" }
                   ]
                 }
-                """));
+                """;
+
+        final var processor = new VulnerabilityMirrorProcessor();
+        processor.process(aConsumerRecord("OSV/GHSA-2cc5-23r7-vc4v", generateBomFromJson(bovJson)).build());
 
         final Vulnerability vuln = qm.getVulnerabilityByVulnId("GITHUB", "GHSA-2cc5-23r7-vc4v");
         assertThat(vuln).isNotNull();
@@ -555,7 +535,7 @@ public void testProcessOsvVuln() throws Exception {
 
     @Test
     public void testProcessVulnWithoutAffects() throws Exception {
-        inputTopic.pipeInput("NVD/CVE-2022-40489", KafkaTestUtil.generateBomFromJson("""
+        final var bovJson = """
                 {
                   "components": [
                     {
@@ -573,7 +553,10 @@ public void testProcessVulnWithoutAffects() throws Exception {
                     }
                   ]
                 }
-                """));
+                """;
+
+        final var processor = new VulnerabilityMirrorProcessor();
+        processor.process(aConsumerRecord("NVD/CVE-2022-40489", generateBomFromJson(bovJson)).build());
 
         final Vulnerability vuln = qm.getVulnerabilityByVulnId("NVD", "CVE-2022-40489");
         assertThat(vuln).isNotNull();
@@ -613,7 +596,7 @@ public void testProcessVulnWithoutAffects() throws Exception {
 
     @Test
     public void testProcessVulnWithUnmatchedAffectsBomRef() throws Exception {
-        inputTopic.pipeInput("NVD/CVE-2022-40489", KafkaTestUtil.generateBomFromJson("""
+        final var bovJson = """
                 {
                   "components": [
                     {
@@ -639,7 +622,10 @@ public void testProcessVulnWithUnmatchedAffectsBomRef() throws Exception {
                     }
                   ]
                 }
-                """));
+                """;
+
+        final var processor = new VulnerabilityMirrorProcessor();
+        processor.process(aConsumerRecord("NVD/CVE-2022-40489", generateBomFromJson(bovJson)).build());
 
         final Vulnerability vuln = qm.getVulnerabilityByVulnId("NVD", "CVE-2022-40489");
         assertThat(vuln).isNotNull();
@@ -679,7 +665,7 @@ public void testProcessVulnWithUnmatchedAffectsBomRef() throws Exception {
 
     @Test
     public void testProcessVulnWithVersConstraints() throws Exception {
-        inputTopic.pipeInput("NVD/CVE-2022-40489", KafkaTestUtil.generateBomFromJson("""
+        final var bovJson = """
                 {
                   "components": [
                     {
@@ -731,7 +717,10 @@ public void testProcessVulnWithVersConstraints() throws Exception {
                     }
                   ]
                 }
-                """));
+                """;
+
+        final var processor = new VulnerabilityMirrorProcessor();
+        processor.process(aConsumerRecord("NVD/CVE-2022-40489", generateBomFromJson(bovJson)).build());
 
         final Vulnerability vuln = qm.getVulnerabilityByVulnId("NVD", "CVE-2022-40489");
         assertThat(vuln).isNotNull();
@@ -935,7 +924,7 @@ public void testProcessVulnWithVersConstraints() throws Exception {
 
     @Test
     public void testProcessVulnWithInvalidCpeOrPurl() throws Exception {
-        inputTopic.pipeInput("NVD/CVE-2022-40489", KafkaTestUtil.generateBomFromJson("""
+        final var bovJson = """
                 {
                   "components": [
                     {
@@ -997,7 +986,10 @@ public void testProcessVulnWithInvalidCpeOrPurl() throws Exception {
                     }
                   ]
                 }
-                """));
+                """;
+
+        final var processor = new VulnerabilityMirrorProcessor();
+        processor.process(aConsumerRecord("NVD/CVE-2022-40489", generateBomFromJson(bovJson)).build());
 
         final Vulnerability vuln = qm.getVulnerabilityByVulnId("NVD", "CVE-2022-40489");
         assertThat(vuln).isNotNull();
@@ -1035,4 +1027,4 @@ public void testProcessVulnWithInvalidCpeOrPurl() throws Exception {
         assertThat(vuln.getVulnerableSoftware()).isEmpty();
     }
 
-}
+}

Original file line number	Diff line number	Diff line change
`@@ -1,55 +1,26 @@`
`1`		`-package org.dependencytrack.event.kafka.streams.processor;`
	`1`	`+package org.dependencytrack.event.kafka.processor;`
`2`	`2`
`3`		`-import org.apache.kafka.common.serialization.Serdes;`
`4`		`-import org.apache.kafka.common.serialization.StringSerializer;`
`5`		`-import org.apache.kafka.streams.StreamsBuilder;`
`6`		`-import org.apache.kafka.streams.TestInputTopic;`
`7`		`-import org.apache.kafka.streams.TopologyTestDriver;`
`8`		`-import org.apache.kafka.streams.kstream.Consumed;`
`9`		`-import org.cyclonedx.proto.v1_4.Bom;`
`10`		`-import org.dependencytrack.PersistenceCapableTest;`
`11`		`-import org.dependencytrack.event.kafka.serialization.KafkaProtobufSerde;`
`12`		`-import org.dependencytrack.event.kafka.serialization.KafkaProtobufSerializer;`
`13`	`3`	`import org.dependencytrack.model.Severity;`
`14`	`4`	`import org.dependencytrack.model.Vulnerability;`
`15`	`5`	`import org.dependencytrack.persistence.CweImporter;`
`16`		`-import org.dependencytrack.util.KafkaTestUtil;`
`17`		`-import org.junit.After;`
`18`	`6`	`import org.junit.Before;`
`19`	`7`	`import org.junit.Test;`
`20`	`8`
`21`	`9`	`import static org.assertj.core.api.Assertions.assertThat;`
	`10`	`+import static org.dependencytrack.util.KafkaTestUtil.generateBomFromJson;`
`22`	`11`
`23`		`-public class MirrorVulnerabilityProcessorTest extends PersistenceCapableTest {`
`24`		`-`
`25`		`- private TopologyTestDriver testDriver;`
`26`		`- private TestInputTopic<String, Bom> inputTopic;`
	`12`	`+public class VulnerabilityMirrorProcessorTest extends AbstractProcessorTest {`
`27`	`13`
`28`	`14`	`@Before`
`29`		`- public void setUp() throws Exception {`
`30`		`- final var streamsBuilder = new StreamsBuilder();`
`31`		`- streamsBuilder`
`32`		`- .stream("input-topic", Consumed`
`33`		`- .with(Serdes.String(), new KafkaProtobufSerde<>(Bom.parser())))`
`34`		`- .process(MirrorVulnerabilityProcessor::new);`
`35`		`-`
`36`		`- testDriver = new TopologyTestDriver(streamsBuilder.build());`
`37`		`- inputTopic = testDriver.createInputTopic("input-topic",`
`38`		`- new StringSerializer(), new KafkaProtobufSerializer<>());`
	`15`	`+ public void before() throws Exception {`
	`16`	`+ super.before();`
`39`	`17`
`40`	`18`	`new CweImporter().processCweDefinitions(); // Required for CWE mapping`
`41`	`19`	`}`
`42`	`20`
`43`		`- @After`
`44`		`- public void tearDown() {`
`45`		`- if (testDriver != null) {`
`46`		`- testDriver.close();`
`47`		`- }`
`48`		`- }`
`49`		`-`
`50`	`21`	`@Test`
`51`	`22`	`public void testProcessNvdVuln() throws Exception {`
`52`		`- inputTopic.pipeInput("NVD/CVE-2022-40489", KafkaTestUtil.generateBomFromJson("""`
	`23`	`+ final var bovJson = """`
`53`	`24`	`{`
`54`	`25`	`"components": [`
`55`	`26`	`{`
`@@ -91,7 +62,10 @@ public void testProcessNvdVuln() throws Exception {`
`91`	`62`	`{ "url": "https://github.com/thinkcmf/thinkcmf/issues/736" }`
`92`	`63`	`]`
`93`	`64`	`}`
`94`		`- """));`
	`65`	`+ """;`
	`66`	`+`
	`67`	`+ final var processor = new VulnerabilityMirrorProcessor();`
	`68`	`+ processor.process(aConsumerRecord("NVD/CVE-2022-40489", generateBomFromJson(bovJson)).build());`
`95`	`69`
`96`	`70`	`final Vulnerability vuln = qm.getVulnerabilityByVulnId("NVD", "CVE-2022-40489");`
`97`	`71`	`assertThat(vuln).isNotNull();`
`@@ -160,7 +134,7 @@ public void testProcessNvdVuln() throws Exception {`
`160`	`134`
`161`	`135`	`@Test`
`162`	`136`	`public void testProcessGitHubVuln() throws Exception {`
`163`		`- inputTopic.pipeInput("GITHUB/GHSA-fxwm-579q-49qq", KafkaTestUtil.generateBomFromJson("""`
	`137`	`+ final var bovJson = """`
`164`	`138`	`{`
`165`	`139`	`"components": [`
`166`	`140`	`{`
`@@ -223,7 +197,10 @@ public void testProcessGitHubVuln() throws Exception {`
`223`	`197`	`{ "url": "https://github.com/advisories/GHSA-fxwm-579q-49qq" }`
`224`	`198`	`]`
`225`	`199`	`}`
`226`		`- """));`
	`200`	`+ """;`
	`201`	`+`
	`202`	`+ final var processor = new VulnerabilityMirrorProcessor();`
	`203`	`+ processor.process(aConsumerRecord("GITHUB/GHSA-fxwm-579q-49qq", generateBomFromJson(bovJson)).build());`
`227`	`204`
`228`	`205`	`final Vulnerability vuln = qm.getVulnerabilityByVulnId("GITHUB", "GHSA-fxwm-579q-49qq");`
`229`	`206`	`assertThat(vuln).isNotNull();`
`@@ -375,7 +352,7 @@ public void testProcessGitHubVuln() throws Exception {`
`375`	`352`
`376`	`353`	`@Test`
`377`	`354`	`public void testProcessOsvVuln() throws Exception {`
`378`		`- inputTopic.pipeInput("OSV/GHSA-2cc5-23r7-vc4v", KafkaTestUtil.generateBomFromJson("""`
	`355`	`+ final var bovJson = """`
`379`	`356`	`{`
`380`	`357`	`"components": [`
`381`	`358`	`{`
`@@ -427,7 +404,10 @@ public void testProcessOsvVuln() throws Exception {`
`427`	`404`	`{ "url": "https://github.com/ratpack/ratpack/blob/29434f7ac6fd4b36a4495429b70f4c8163100332/ratpack-session/src/main/java/ratpack/session/clientside/ClientSideSessionConfig.java#L29" }`
`428`	`405`	`]`
`429`	`406`	`}`
`430`		`- """));`
	`407`	`+ """;`
	`408`	`+`
	`409`	`+ final var processor = new VulnerabilityMirrorProcessor();`
	`410`	`+ processor.process(aConsumerRecord("OSV/GHSA-2cc5-23r7-vc4v", generateBomFromJson(bovJson)).build());`
`431`	`411`
`432`	`412`	`final Vulnerability vuln = qm.getVulnerabilityByVulnId("GITHUB", "GHSA-2cc5-23r7-vc4v");`
`433`	`413`	`assertThat(vuln).isNotNull();`
`@@ -555,7 +535,7 @@ public void testProcessOsvVuln() throws Exception {`
`555`	`535`
`556`	`536`	`@Test`
`557`	`537`	`public void testProcessVulnWithoutAffects() throws Exception {`
`558`		`- inputTopic.pipeInput("NVD/CVE-2022-40489", KafkaTestUtil.generateBomFromJson("""`
	`538`	`+ final var bovJson = """`
`559`	`539`	`{`
`560`	`540`	`"components": [`
`561`	`541`	`{`
`@@ -573,7 +553,10 @@ public void testProcessVulnWithoutAffects() throws Exception {`
`573`	`553`	`}`
`574`	`554`	`]`
`575`	`555`	`}`
`576`		`- """));`
	`556`	`+ """;`
	`557`	`+`
	`558`	`+ final var processor = new VulnerabilityMirrorProcessor();`
	`559`	`+ processor.process(aConsumerRecord("NVD/CVE-2022-40489", generateBomFromJson(bovJson)).build());`
`577`	`560`
`578`	`561`	`final Vulnerability vuln = qm.getVulnerabilityByVulnId("NVD", "CVE-2022-40489");`
`579`	`562`	`assertThat(vuln).isNotNull();`
`@@ -613,7 +596,7 @@ public void testProcessVulnWithoutAffects() throws Exception {`
`613`	`596`
`614`	`597`	`@Test`
`615`	`598`	`public void testProcessVulnWithUnmatchedAffectsBomRef() throws Exception {`
`616`		`- inputTopic.pipeInput("NVD/CVE-2022-40489", KafkaTestUtil.generateBomFromJson("""`
	`599`	`+ final var bovJson = """`
`617`	`600`	`{`
`618`	`601`	`"components": [`
`619`	`602`	`{`
`@@ -639,7 +622,10 @@ public void testProcessVulnWithUnmatchedAffectsBomRef() throws Exception {`
`639`	`622`	`}`
`640`	`623`	`]`
`641`	`624`	`}`
`642`		`- """));`
	`625`	`+ """;`
	`626`	`+`
	`627`	`+ final var processor = new VulnerabilityMirrorProcessor();`
	`628`	`+ processor.process(aConsumerRecord("NVD/CVE-2022-40489", generateBomFromJson(bovJson)).build());`
`643`	`629`
`644`	`630`	`final Vulnerability vuln = qm.getVulnerabilityByVulnId("NVD", "CVE-2022-40489");`
`645`	`631`	`assertThat(vuln).isNotNull();`
`@@ -679,7 +665,7 @@ public void testProcessVulnWithUnmatchedAffectsBomRef() throws Exception {`
`679`	`665`
`680`	`666`	`@Test`
`681`	`667`	`public void testProcessVulnWithVersConstraints() throws Exception {`
`682`		`- inputTopic.pipeInput("NVD/CVE-2022-40489", KafkaTestUtil.generateBomFromJson("""`
	`668`	`+ final var bovJson = """`
`683`	`669`	`{`
`684`	`670`	`"components": [`
`685`	`671`	`{`
`@@ -731,7 +717,10 @@ public void testProcessVulnWithVersConstraints() throws Exception {`
`731`	`717`	`}`
`732`	`718`	`]`
`733`	`719`	`}`
`734`		`- """));`
	`720`	`+ """;`
	`721`	`+`
	`722`	`+ final var processor = new VulnerabilityMirrorProcessor();`
	`723`	`+ processor.process(aConsumerRecord("NVD/CVE-2022-40489", generateBomFromJson(bovJson)).build());`
`735`	`724`
`736`	`725`	`final Vulnerability vuln = qm.getVulnerabilityByVulnId("NVD", "CVE-2022-40489");`
`737`	`726`	`assertThat(vuln).isNotNull();`
`@@ -935,7 +924,7 @@ public void testProcessVulnWithVersConstraints() throws Exception {`
`935`	`924`
`936`	`925`	`@Test`
`937`	`926`	`public void testProcessVulnWithInvalidCpeOrPurl() throws Exception {`
`938`		`- inputTopic.pipeInput("NVD/CVE-2022-40489", KafkaTestUtil.generateBomFromJson("""`
	`927`	`+ final var bovJson = """`
`939`	`928`	`{`
`940`	`929`	`"components": [`
`941`	`930`	`{`
`@@ -997,7 +986,10 @@ public void testProcessVulnWithInvalidCpeOrPurl() throws Exception {`
`997`	`986`	`}`
`998`	`987`	`]`
`999`	`988`	`}`
`1000`		`- """));`
	`989`	`+ """;`
	`990`	`+`
	`991`	`+ final var processor = new VulnerabilityMirrorProcessor();`
	`992`	`+ processor.process(aConsumerRecord("NVD/CVE-2022-40489", generateBomFromJson(bovJson)).build());`
`1001`	`993`
`1002`	`994`	`final Vulnerability vuln = qm.getVulnerabilityByVulnId("NVD", "CVE-2022-40489");`
`1003`	`995`	`assertThat(vuln).isNotNull();`
`@@ -1035,4 +1027,4 @@ public void testProcessVulnWithInvalidCpeOrPurl() throws Exception {`
`1035`	`1027`	`assertThat(vuln.getVulnerableSoftware()).isEmpty();`
`1036`	`1028`	`}`
`1037`	`1029`
`1038`		`-}`
	`1030`	`+}`