|
26 | 26 | import io.quarkus.runtime.ShutdownEvent;
|
27 | 27 | import io.quarkus.runtime.annotations.RegisterForReflection;
|
28 | 28 | import io.smallrye.common.annotation.Identifier;
|
29 |
| -import jakarta.enterprise.context.ApplicationScoped; |
30 |
| -import jakarta.enterprise.event.Observes; |
31 |
| -import jakarta.enterprise.inject.Produces; |
32 | 29 | import org.apache.kafka.clients.consumer.ConsumerConfig;
|
33 | 30 | import org.apache.kafka.clients.consumer.KafkaConsumer;
|
34 | 31 | import org.apache.kafka.clients.consumer.internals.AsyncKafkaConsumer;
|
| 32 | +import org.apache.kafka.clients.consumer.internals.ClassicKafkaConsumer; |
35 | 33 | import org.apache.kafka.clients.consumer.internals.ConsumerCoordinator;
|
36 |
| -import org.apache.kafka.clients.consumer.internals.LegacyKafkaConsumer; |
37 | 34 | import org.apache.kafka.common.serialization.StringDeserializer;
|
38 | 35 | import org.dependencytrack.notification.serialization.NotificationKafkaProtobufDeserializer;
|
39 | 36 | import org.dependencytrack.proto.notification.v1.Notification;
|
40 | 37 | import org.eclipse.microprofile.config.ConfigProvider;
|
41 | 38 |
|
| 39 | +import jakarta.enterprise.context.ApplicationScoped; |
| 40 | +import jakarta.enterprise.event.Observes; |
| 41 | +import jakarta.enterprise.inject.Produces; |
42 | 42 | import java.time.Duration;
|
43 | 43 | import java.util.HashMap;
|
44 | 44 | import java.util.Map;
|
|
52 | 52 | // https://github.com/confluentinc/parallel-consumer/pull/762
|
53 | 53 | KafkaConsumer.class,
|
54 | 54 | AsyncKafkaConsumer.class,
|
55 |
| - LegacyKafkaConsumer.class, |
| 55 | + ClassicKafkaConsumer.class, |
56 | 56 | ConsumerCoordinator.class
|
57 | 57 | })
|
58 | 58 | class ParallelConsumerConfiguration {
|
@@ -91,6 +91,7 @@ private static ParallelStreamProcessor<String, Notification> createParallelConsu
|
91 | 91 | final var parallelConsumerOptions = ParallelConsumerOptions.<String, Notification>builder()
|
92 | 92 | .consumer(consumer)
|
93 | 93 | .maxConcurrency(parallelConsumerConfig.maxConcurrency())
|
| 94 | + .ignoreReflectiveAccessExceptionsForAutoCommitDisabledCheck(true) |
94 | 95 | .ordering(ProcessingOrder.KEY)
|
95 | 96 | .retryDelayProvider(recordCtx -> {
|
96 | 97 | final long delayMillis = RetryConfig
|
|
0 commit comments