diff --git a/.github/project.yml b/.github/project.yml index 33eca9b..3cc02cf 100644 --- a/.github/project.yml +++ b/.github/project.yml @@ -1,3 +1,3 @@ release: - current-version: "3.0.1" - next-version: "3.0.0-SNAPSHOT" + current-version: "4.0.1" + next-version: "4.0.0-SNAPSHOT" diff --git a/api/pom.xml b/api/pom.xml index e167114..b2cf9e4 100644 --- a/api/pom.xml +++ b/api/pom.xml @@ -4,7 +4,7 @@ io.quarkiverse.kafkastreamsprocessor quarkus-kafka-streams-processor-parent - 3.0.0-SNAPSHOT + 4.0.0-SNAPSHOT quarkus-kafka-streams-processor-api diff --git a/api/src/main/java/io/quarkiverse/kafkastreamsprocessor/api/decorator/processor/AbstractProcessorDecorator.java b/api/src/main/java/io/quarkiverse/kafkastreamsprocessor/api/decorator/processor/AbstractProcessorDecorator.java new file mode 100644 index 0000000..fdc99b4 --- /dev/null +++ b/api/src/main/java/io/quarkiverse/kafkastreamsprocessor/api/decorator/processor/AbstractProcessorDecorator.java @@ -0,0 +1,55 @@ +/*- + * #%L + * Quarkus Kafka Streams Processor + * %% + * Copyright (C) 2024 Amadeus s.a.s. + * %% + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * #L% + */ +package io.quarkiverse.kafkastreamsprocessor.api.decorator.processor; + +import org.apache.kafka.streams.processor.api.Processor; + +import lombok.Getter; +import lombok.Setter; +import lombok.experimental.Delegate; + +/** + * Base class for all processor decorators. + *

+ * If a decorator does not implement this abstract class, it will not be found by the + * KafkaClientSuppliedDecorator for composition. + *

+ *

+ * We remove the generic declaration from {@link Processor} because ArC complains about generics on class declaration of + * a bean. + *

+ *

+ * Class introduced in 2.0, for compatibility with Quarkus 3.8 random failure to start when using custom processor + * decorators. + *

+ * + * @deprecated It will be removed in 3.0, with the integration of Quarkus 3.15 where we will be able to go back to pure + * CDI decorators. + */ +@Deprecated(forRemoval = true, since = "2.0") +public abstract class AbstractProcessorDecorator implements Processor { + /** + * The decorated processor, holding either the next decorator layer or the final processor. + */ + @Delegate + @Getter + @Setter + private Processor delegate; +} diff --git a/bom/application/pom.xml b/bom/application/pom.xml index 24f637a..3a8aa55 100644 --- a/bom/application/pom.xml +++ b/bom/application/pom.xml @@ -4,12 +4,12 @@ io.quarkiverse.kafkastreamsprocessor quarkus-kafka-streams-processor-bom-parent - 3.0.0-SNAPSHOT + 4.0.0-SNAPSHOT quarkus-kafka-streams-processor-bom pom - 3.0.0-SNAPSHOT + 4.0.0-SNAPSHOT diff --git a/bom/pom.xml b/bom/pom.xml index d35e6aa..a3ebca8 100644 --- a/bom/pom.xml +++ b/bom/pom.xml @@ -4,7 +4,7 @@ io.quarkiverse.kafkastreamsprocessor quarkus-kafka-streams-processor-parent - 3.0.0-SNAPSHOT + 4.0.0-SNAPSHOT quarkus-kafka-streams-processor-bom-parent pom diff --git a/bom/test/pom.xml b/bom/test/pom.xml index 06d1696..89a79b4 100644 --- a/bom/test/pom.xml +++ b/bom/test/pom.xml @@ -4,7 +4,7 @@ io.quarkiverse.kafkastreamsprocessor quarkus-kafka-streams-processor-bom-parent - 3.0.0-SNAPSHOT + 4.0.0-SNAPSHOT quarkus-kafka-streams-processor-test-bom pom diff --git a/deployment/pom.xml b/deployment/pom.xml index 7a08f9b..81cc8d8 100644 --- a/deployment/pom.xml +++ b/deployment/pom.xml @@ -4,7 +4,7 @@ io.quarkiverse.kafkastreamsprocessor quarkus-kafka-streams-processor-parent - 3.0.0-SNAPSHOT + 4.0.0-SNAPSHOT quarkus-kafka-streams-processor-deployment diff --git a/docs/modules/ROOT/pages/includes/attributes.adoc b/docs/modules/ROOT/pages/includes/attributes.adoc index ca824b3..9b6838f 100644 --- a/docs/modules/ROOT/pages/includes/attributes.adoc +++ b/docs/modules/ROOT/pages/includes/attributes.adoc @@ -1,3 +1,3 @@ -:project-version: 2.0.1 +:project-version: 3.0.1 :examples-dir: ./../examples/ \ No newline at end of file diff --git a/docs/modules/ROOT/pages/index.adoc b/docs/modules/ROOT/pages/index.adoc index 308cbe6..a89dc88 100644 --- a/docs/modules/ROOT/pages/index.adoc +++ b/docs/modules/ROOT/pages/index.adoc @@ -690,41 +690,25 @@ The extension proposes some capabilities to customize more finely the behaviour === Processor decorator The following decoration layer is already extensively used in this extension's source code and allows to use composition around the main processor class you have to define. -Depending on the version of Quarkus you are using, the pattern differs: +The pattern to implement is: -.ProcessorDecorator.java with Quarkus 3.2 or 3.11.0+ [source,java] ---- -@Decorator // <1> +@Dependent // <1> @Priority(150) // <2> -public class ProcessorDecorator implements Processor { // <3> - @lombok.experimental.Delegate(excludes = Excludes.class) - private final Processor delegate; // <4> - - private final MyBean bean; - - @Inject - public TracingDecorator(@Delegate Processor delegate, MyBean bean) { // <5> - this.delegate = delegate; - this.bean = bean; - } - +public class ProcessorDecorator extends AbstractProcessorDecorator { // <3> @Override - public void process(Record record) { // <6> + public void process(Record record) { // <4> // use bean before - delegate.process(record); + getDelegate().process(record); // use bean after } - - private interface Excludes { - void process(Record record); - } } ---- -<1> Decorator annotation to profit from the {cdi-spec}/cdi-spec.html#decorators[decorator] feature of CDI -<2> Force the instantiation of the decorator with the Priority annotation. -Indeed, otherwise the decorator is not taken into account by Quarkus. +<1> We have to mark the bean `Dependent` so it is instantiated at every use. +Indeed, `KStreamProcessorSupplier` needs to return a new `Processor` instance everytime it is called, by Kafka Streams' specification. +<2> We add a `Priority`, to control exactly the order of decorators. The priority is to be set based on the priorities of the existing decorators which are: + .ProcessorDecoratorPriorities.java @@ -760,49 +744,12 @@ The priority is to be set based on the priorities of the existing decorators whi */ public static final int RETRY = 600; ---- -<3> The decorator should have the same generics declaration `` as the `Processor` interface that it implements -<4> Delegate reference to use when decorating methods. -It is annotated with lombok's https://projectlombok.org/features/experimental/Delegate[Delegate] annotation to generate passthrough decorated methods that this Decorator class won't decorate. -The selection is done through a blacklist of method signatures gathered in a private `Excludes` interface declared at the end of the class. -<5> Injection constructor which must have a delegate argument annotated with the `Delegate` annotation from CDI. -You can also, as a regular CDI bean, inject any another CDI bean reference to be used in this decorator. -<6> Example of decorated method, here the main `process` method of `Processor` API of Kafka Streams. - -.ProcessorDecorator.java for Quarkus 3.8 -> 3.10 -[source,java] ----- -@Dependent // <1> -@Priority(150) // <2> -public class ProcessorDecorator extends AbstractProcessorDecorator { // <3> - @Override - public void process(Record record) { // <4> - // use bean before - getDelegate().process(record); - // use bean after - } -} ----- - -<1> We have to mark the bean `Dependent` so it is instantiated at every use. -Indeed, `KStreamProcessorSupplier` needs to return a new `Processor` instance everytime it is called, by Kafka Streams' specification. -<2> We add a `Priority`, with same pattern as a CDI decorator. <3> We remove the generic types from the class signature, because CDI does not like generics in beans. <4> Example of override of process method and call to underlying decorator. Such a decorator will automatically been taken into account by CDI. The priority will control at which point your decorator will be called among all other decorators. -[CAUTION] -==== -We noticed with a new integration-test that is using a custom serde, that usage of custom CDI `Decorator` is causing microservices to randomly crash at startup. -This happens for specific versions of Quarkus. -Known impacted versions are 3.8.x, 3.9.x and 3.10.x. -The 3.2 LTS and upcoming 3.15 LTS versions do not suffer from this symptom. -The **only** solution found was to remove usage of `@Decorator` for `Processor` decorators for microservices based on Quarkus 3.8 LTS. -This change will be reverted in quarkus-kafka-streams-processor 3.0. -This is probably the https://github.com/quarkusio/quarkus/pull/41258[PR] on Quarkus side that has fixed the issue in Quarkus 3.11. -==== - === Producer interceptor Kafka Streams already has the concept of a https://kafka.apache.org/36/javadoc/org/apache/kafka/clients/producer/ProducerInterceptor.html[ProducerInterceptor]. @@ -919,4 +866,4 @@ include::includes/kafka-streams-processor-configuration-keys.adoc[] == Configuration from other extension -include::includes/quarkus-other-extension-configurations.adoc[] \ No newline at end of file +include::includes/quarkus-other-extension-configurations.adoc[] diff --git a/docs/pom.xml b/docs/pom.xml index 9ee7d56..bdec31c 100644 --- a/docs/pom.xml +++ b/docs/pom.xml @@ -4,7 +4,7 @@ io.quarkiverse.kafkastreamsprocessor quarkus-kafka-streams-processor-parent - 3.0.0-SNAPSHOT + 4.0.0-SNAPSHOT ../pom.xml diff --git a/impl/pom.xml b/impl/pom.xml index 7240c14..c626054 100644 --- a/impl/pom.xml +++ b/impl/pom.xml @@ -4,7 +4,7 @@ io.quarkiverse.kafkastreamsprocessor quarkus-kafka-streams-processor-parent - 3.0.0-SNAPSHOT + 4.0.0-SNAPSHOT quarkus-kafka-streams-processor-impl diff --git a/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/KStreamProcessorSupplier.java b/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/KStreamProcessorSupplier.java index ec09ab4..f3af0b7 100644 --- a/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/KStreamProcessorSupplier.java +++ b/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/KStreamProcessorSupplier.java @@ -20,15 +20,16 @@ package io.quarkiverse.kafkastreamsprocessor.impl; import java.lang.annotation.Annotation; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; import java.util.Optional; -import java.util.Set; import java.util.stream.Collectors; import jakarta.enterprise.context.ApplicationScoped; import jakarta.enterprise.context.Dependent; import jakarta.enterprise.context.RequestScoped; import jakarta.enterprise.inject.Instance; -import jakarta.enterprise.inject.spi.Bean; import jakarta.enterprise.inject.spi.BeanManager; import jakarta.inject.Inject; import jakarta.inject.Singleton; @@ -36,6 +37,7 @@ import org.apache.kafka.streams.processor.api.Processor; import org.apache.kafka.streams.processor.api.ProcessorSupplier; +import io.quarkiverse.kafkastreamsprocessor.api.decorator.processor.AbstractProcessorDecorator; import lombok.extern.slf4j.Slf4j; /** @@ -60,6 +62,8 @@ public class KStreamProcessorSupplier implements Processor */ private final Instance> adapterInstances; + private final Instance processorDecorators; + /** * Injection constructor. * @@ -76,17 +80,20 @@ public class KStreamProcessorSupplier implements Processor @Inject public KStreamProcessorSupplier(Instance> kafka3BeanInstances, Instance> beanInstances, - Instance> adapterInstances, BeanManager beanManager) { + Instance> adapterInstances, BeanManager beanManager, + Instance processorDecorators) { this.kafka3BeanInstances = kafka3BeanInstances; this.beanInstances = beanInstances; this.adapterInstances = adapterInstances; + this.processorDecorators = processorDecorators; + + List processorDecoratorNames = new ArrayList<>(processorDecorators.stream() + .map(Object::getClass) + .map(Class::getName) + .collect(Collectors.toUnmodifiableList())); + Collections.reverse(processorDecoratorNames); + log.info("Configured Processor decorators are in order: {}", String.join(", ", processorDecoratorNames)); - log.info("Configured Processor decorators are in order: {}", - beanManager.resolveDecorators(Set.of(Processor.class)) - .stream() - .map(Bean::getBeanClass) - .map(Class::getName) - .collect(Collectors.joining(", "))); } /** @@ -131,7 +138,16 @@ public Processor get() { "Processors cannot have a scope other than @Dependant, since KafkaStreams implementation classes are not thread-safe"); } - return (Processor) processor; + return wrapProcessor((Processor) processor); + } + + private Processor wrapProcessor(Processor processor) { + Processor wrappedProcessor = processor; + for (AbstractProcessorDecorator decorator : processorDecorators) { + decorator.setDelegate(wrappedProcessor); + wrappedProcessor = decorator; + } + return wrappedProcessor; } private static boolean hasAnnotation(Object bean, Class annotation) { diff --git a/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/CdiRequestContextDecorator.java b/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/CdiRequestContextDecorator.java index fc92484..7781526 100644 --- a/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/CdiRequestContextDecorator.java +++ b/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/CdiRequestContextDecorator.java @@ -20,17 +20,16 @@ package io.quarkiverse.kafkastreamsprocessor.impl.decorator.processor; import jakarta.annotation.Priority; -import jakarta.decorator.Decorator; +import jakarta.enterprise.context.Dependent; import jakarta.inject.Inject; import org.apache.kafka.streams.processor.api.Processor; import org.apache.kafka.streams.processor.api.Record; +import io.quarkiverse.kafkastreamsprocessor.api.decorator.processor.AbstractProcessorDecorator; import io.quarkiverse.kafkastreamsprocessor.api.decorator.processor.ProcessorDecoratorPriorities; import io.quarkus.arc.Arc; import io.quarkus.arc.ArcContainer; -import lombok.AccessLevel; -import lombok.RequiredArgsConstructor; /** * This class is responsible to manage the lifecycle of {@link jakarta.enterprise.context.RequestScoped} beans. It @@ -41,16 +40,11 @@ *

* Warning: "Quarkus Tests" Junit extension is already managing the request scope on its own. */ -@Decorator +//@Decorator +@Dependent @Priority(ProcessorDecoratorPriorities.CDI_REQUEST_SCOPE) -@RequiredArgsConstructor(access = AccessLevel.MODULE) -public class CdiRequestContextDecorator implements Processor { - /** - * Injection point for composition - */ - @lombok.experimental.Delegate(excludes = Excludes.class) - private final Processor delegate; - +//@RequiredArgsConstructor(access = AccessLevel.MODULE) +public class CdiRequestContextDecorator extends AbstractProcessorDecorator { /** * The container object from Arc to inquire on request contextualization availability and activation */ @@ -58,13 +52,14 @@ public class CdiRequestContextDecorator implements Process /** * Constructor for injection of the delegate. - * - * @param delegate - * injection point for composition */ @Inject - public CdiRequestContextDecorator(@jakarta.decorator.Delegate Processor delegate) { - this(delegate, Arc.container()); + public CdiRequestContextDecorator() { + this(Arc.container()); + } + + public CdiRequestContextDecorator(ArcContainer container) { + this.container = container; } /** @@ -74,20 +69,16 @@ public CdiRequestContextDecorator(@jakarta.decorator.Delegate Processor record) { + public void process(Record record) { if (container.requestContext().isActive()) { - delegate.process(record); + getDelegate().process(record); } else { container.requestContext().activate(); try { - delegate.process(record); + getDelegate().process(record); } finally { container.requestContext().terminate(); } } } - - private interface Excludes { - void process(Record record); - } } diff --git a/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/DlqDecorator.java b/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/DlqDecorator.java index a13b62d..f3a6eaf 100644 --- a/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/DlqDecorator.java +++ b/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/DlqDecorator.java @@ -23,8 +23,8 @@ import java.util.Set; import jakarta.annotation.Priority; -import jakarta.decorator.Decorator; import jakarta.decorator.Delegate; +import jakarta.enterprise.context.Dependent; import jakarta.inject.Inject; import org.apache.kafka.common.KafkaException; @@ -37,6 +37,7 @@ import org.apache.kafka.streams.processor.api.RecordMetadata; import org.apache.kafka.streams.processor.internals.InternalProcessorContext; +import io.quarkiverse.kafkastreamsprocessor.api.decorator.processor.AbstractProcessorDecorator; import io.quarkiverse.kafkastreamsprocessor.api.decorator.processor.ProcessorDecoratorPriorities; import io.quarkiverse.kafkastreamsprocessor.impl.TopologyProducer; import io.quarkiverse.kafkastreamsprocessor.impl.errors.DlqMetadataHandler; @@ -53,14 +54,10 @@ * Uses a dead-letter sink from the topology, rather than a raw producer, to benefit from the same KStreams guarantees * (at least once / exactly once). */ -@Decorator +//@Decorator @Priority(ProcessorDecoratorPriorities.DLQ) -public class DlqDecorator implements Processor { - /** - * Inject point for composition - */ - @lombok.experimental.Delegate(excludes = Excludes.class) - private final Processor delegate; +@Dependent +public class DlqDecorator extends AbstractProcessorDecorator { /** * A set of sink names that are involved in the business logic. @@ -89,11 +86,10 @@ public class DlqDecorator implements Processor context; + private ProcessorContext context; - DlqDecorator(Processor delegate, Set functionalSinks, DlqMetadataHandler dlqMetadataHandler, + DlqDecorator(Set functionalSinks, DlqMetadataHandler dlqMetadataHandler, KafkaStreamsProcessorMetrics metrics, boolean activated) { - this.delegate = delegate; this.functionalSinks = functionalSinks; this.dlqMetadataHandler = dlqMetadataHandler; this.metrics = metrics; @@ -103,8 +99,6 @@ public class DlqDecorator implements Processor implements Processor delegate, + public DlqDecorator( SinkToTopicMappingBuilder sinkToTopicMappingBuilder, DlqMetadataHandler dlqMetadataHandler, KafkaStreamsProcessorMetrics metrics, KStreamsProcessorConfig kStreamsProcessorConfig) { // NOSONAR Optional with microprofile-config - this(delegate, sinkToTopicMappingBuilder.sinkToTopicMapping().keySet(), dlqMetadataHandler, metrics, + this(sinkToTopicMappingBuilder.sinkToTopicMapping().keySet(), dlqMetadataHandler, metrics, ErrorHandlingStrategy.shouldSendToDlq(kStreamsProcessorConfig.errorStrategy(), kStreamsProcessorConfig.dlq().topic())); } @@ -135,12 +129,12 @@ public DlqDecorator(@Delegate Processor delegate, * {@inheritDoc} */ @Override - public void init(final ProcessorContext context) { + public void init(final ProcessorContext context) { if (activated) { - this.context = new DlqProcessorContextDecorator<>((InternalProcessorContext) context, functionalSinks); - delegate.init(this.context); + this.context = new DlqProcessorContextDecorator<>((InternalProcessorContext) context, functionalSinks); + getDelegate().init(this.context); } else { - delegate.init(context); + getDelegate().init(context); } } @@ -154,10 +148,10 @@ public void init(final ProcessorContext context) { * {@inheritDoc} */ @Override - public void process(Record record) { + public void process(Record record) { if (activated) { try { - delegate.process(record); + getDelegate().process(record); } catch (KafkaException e) { // Do not forward to DLQ throw e; @@ -166,14 +160,14 @@ public void process(Record record) { if (recordMetadata.isPresent()) { dlqMetadataHandler.addMetadata(record.headers(), recordMetadata.get().topic(), recordMetadata.get().partition(), e); - context.forward((Record) record, TopologyProducer.DLQ_SINK_NAME); + context.forward(record, TopologyProducer.DLQ_SINK_NAME); // Re-throw so the exception gets logged metrics.microserviceDlqSentCounter().increment(); throw e; } } } else { - delegate.process(record); + getDelegate().process(record); } } diff --git a/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/MetricsDecorator.java b/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/MetricsDecorator.java index 0e73b00..3f44f62 100644 --- a/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/MetricsDecorator.java +++ b/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/MetricsDecorator.java @@ -21,27 +21,23 @@ import jakarta.annotation.Priority; import jakarta.decorator.Decorator; -import jakarta.decorator.Delegate; +import jakarta.enterprise.context.Dependent; import jakarta.inject.Inject; import org.apache.kafka.streams.processor.api.Processor; import org.apache.kafka.streams.processor.api.Record; +import io.quarkiverse.kafkastreamsprocessor.api.decorator.processor.AbstractProcessorDecorator; import io.quarkiverse.kafkastreamsprocessor.api.decorator.processor.ProcessorDecoratorPriorities; import io.quarkiverse.kafkastreamsprocessor.impl.metrics.KafkaStreamsProcessorMetrics; /** * Decorator to enrich Kafka Streams metrics with a counter of exception raised by {@link Processor#process(Record)}. */ -@Decorator +//@Decorator @Priority(ProcessorDecoratorPriorities.METRICS) -public class MetricsDecorator implements Processor { - /** - * Injection point for composition. - */ - @lombok.experimental.Delegate(excludes = Excludes.class) - private final Processor delegate; - +@Dependent +public class MetricsDecorator extends AbstractProcessorDecorator { /** * Counter of exception raised by {@link Processor#process(Record)}. */ @@ -50,15 +46,11 @@ public class MetricsDecorator implements Processor delegate, - KafkaStreamsProcessorMetrics metrics) { - this.delegate = delegate; + public MetricsDecorator(KafkaStreamsProcessorMetrics metrics) { this.metrics = metrics; } @@ -70,9 +62,9 @@ public MetricsDecorator(@Delegate Processor delegate, * {@inheritDoc} */ @Override - public void process(Record record) { + public void process(Record record) { try { - delegate.process(record); + getDelegate().process(record); } catch (Exception e) { // NOSONAR: Catching any error metrics.processorErrorCounter().increment(); throw e; diff --git a/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/PunctuatorDecorationProcessorDecorator.java b/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/PunctuatorDecorationProcessorDecorator.java index 5e1c1e0..cc4fa6c 100644 --- a/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/PunctuatorDecorationProcessorDecorator.java +++ b/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/PunctuatorDecorationProcessorDecorator.java @@ -24,6 +24,7 @@ import jakarta.annotation.Priority; import jakarta.decorator.Decorator; import jakarta.decorator.Delegate; +import jakarta.enterprise.context.Dependent; import jakarta.enterprise.inject.Instance; import jakarta.inject.Inject; @@ -34,6 +35,7 @@ import org.apache.kafka.streams.processor.api.ProcessorContext; import org.apache.kafka.streams.processor.internals.InternalProcessorContext; +import io.quarkiverse.kafkastreamsprocessor.api.decorator.processor.AbstractProcessorDecorator; import io.quarkiverse.kafkastreamsprocessor.api.decorator.processor.ProcessorDecoratorPriorities; import io.quarkiverse.kafkastreamsprocessor.api.decorator.punctuator.DecoratedPunctuator; import lombok.RequiredArgsConstructor; @@ -44,15 +46,10 @@ * * @see PunctuatorDecorationProcessorContextDecorator */ -@Decorator +//@Decorator @Priority(ProcessorDecoratorPriorities.PUNCTUATOR_DECORATION) -public class PunctuatorDecorationProcessorDecorator implements Processor { - /** - * Injection point for composition - */ - @lombok.experimental.Delegate(excludes = Excludes.class) - private final Processor delegate; - +@Dependent +public class PunctuatorDecorationProcessorDecorator extends AbstractProcessorDecorator { /** * List of all the {@link Punctuator} decorators defined in this library and potential extensions made with the API. */ @@ -61,16 +58,13 @@ public class PunctuatorDecorationProcessorDecorator implem /** * Injection constructor. * - * @param delegate - * injection point for composition * @param decoratedPunctuators * the list of all {@link Punctuator} decorators defined in this library and potential extensions made with * the API. */ @Inject - public PunctuatorDecorationProcessorDecorator(@Delegate Processor delegate, + public PunctuatorDecorationProcessorDecorator( Instance decoratedPunctuators) { - this.delegate = delegate; this.decoratedPunctuators = decoratedPunctuators; } @@ -84,8 +78,8 @@ public PunctuatorDecorationProcessorDecorator(@Delegate Processor context) { - delegate.init(new PunctuatorDecorationProcessorContextDecorator<>((InternalProcessorContext) context, + public void init(ProcessorContext context) { + getDelegate().init(new PunctuatorDecorationProcessorContextDecorator<>((InternalProcessorContext) context, decoratedPunctuators)); } diff --git a/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/RetryDecorator.java b/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/RetryDecorator.java index 92d75a7..5b79a55 100644 --- a/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/RetryDecorator.java +++ b/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/RetryDecorator.java @@ -21,13 +21,14 @@ import jakarta.annotation.Priority; import jakarta.decorator.Decorator; -import jakarta.decorator.Delegate; +import jakarta.enterprise.context.Dependent; import jakarta.inject.Inject; import org.apache.kafka.streams.processor.api.Processor; import org.apache.kafka.streams.processor.api.Record; import org.eclipse.microprofile.faulttolerance.Retry; +import io.quarkiverse.kafkastreamsprocessor.api.decorator.processor.AbstractProcessorDecorator; import io.quarkiverse.kafkastreamsprocessor.api.decorator.processor.ProcessorDecoratorPriorities; import lombok.extern.slf4j.Slf4j; @@ -35,15 +36,10 @@ * Decorate a {@link Processor#process} with the {@link Retry} fault tolerance annotation. */ @Slf4j -@Decorator +//@Decorator @Priority(ProcessorDecoratorPriorities.RETRY) -public class RetryDecorator implements Processor { - /** - * Injection point for composition - */ - @lombok.experimental.Delegate(excludes = Excludes.class) - private final Processor delegate; - +@Dependent +public class RetryDecorator extends AbstractProcessorDecorator { /** * The delegate object that has the processor method with the {@link Retry} annotation. *

@@ -58,15 +54,12 @@ public class RetryDecorator implements Processor delegate, + public RetryDecorator( RetryDecoratorDelegate retryDecoratorDelegate) { - this.delegate = delegate; this.retryDecoratorDelegate = retryDecoratorDelegate; } @@ -78,9 +71,9 @@ public RetryDecorator(@Delegate Processor delegate, * {@inheritDoc} */ @Override - public void process(Record record) { + public void process(Record record) { try { - retryDecoratorDelegate.retryableProcess(delegate, record); + retryDecoratorDelegate.retryableProcess(getDelegate(), record); } catch (RuntimeException e) { log.info("An exception that has been raised by the processor will not be retried.\n" + "Possible causes:\n" diff --git a/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/TracingDecorator.java b/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/TracingDecorator.java index 5c05333..5597a68 100644 --- a/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/TracingDecorator.java +++ b/impl/src/main/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/TracingDecorator.java @@ -31,7 +31,7 @@ import jakarta.annotation.Priority; import jakarta.decorator.Decorator; -import jakarta.decorator.Delegate; +import jakarta.enterprise.context.Dependent; import jakarta.inject.Inject; import org.apache.kafka.common.KafkaException; @@ -54,13 +54,12 @@ import io.opentelemetry.context.Context; import io.opentelemetry.context.Scope; import io.opentelemetry.context.propagation.TextMapPropagator; +import io.quarkiverse.kafkastreamsprocessor.api.decorator.processor.AbstractProcessorDecorator; import io.quarkiverse.kafkastreamsprocessor.api.decorator.processor.ProcessorDecoratorPriorities; import io.quarkiverse.kafkastreamsprocessor.impl.configuration.TopologyConfigurationImpl; import io.quarkiverse.kafkastreamsprocessor.impl.protocol.KafkaStreamsProcessorHeaders; import io.quarkiverse.kafkastreamsprocessor.propagation.KafkaTextMapGetter; import io.quarkiverse.kafkastreamsprocessor.propagation.KafkaTextMapSetter; -import lombok.AccessLevel; -import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; /** @@ -72,16 +71,10 @@ * {@link org.apache.kafka.clients.consumer.ConsumerInterceptor}, which executes on the polling thread. */ @Slf4j -@Decorator +//@Decorator @Priority(ProcessorDecoratorPriorities.TRACING) -@RequiredArgsConstructor(access = AccessLevel.MODULE) -public class TracingDecorator implements Processor { - /** - * Injection point for composition - */ - @lombok.experimental.Delegate(excludes = Excludes.class) - private final Processor delegate; - +@Dependent +public class TracingDecorator extends AbstractProcessorDecorator { /** * The {@link OpenTelemetry} configured by Quarkus */ @@ -115,13 +108,11 @@ public class TracingDecorator implements Processor context; + private ProcessorContext context; /** * Injection constructor. * - * @param delegate - * injection point for composition * @param openTelemetry * The {@link OpenTelemetry} configured by Quarkus * @param textMapGetter @@ -134,14 +125,24 @@ public class TracingDecorator implements Processor delegate, OpenTelemetry openTelemetry, + public TracingDecorator(OpenTelemetry openTelemetry, KafkaTextMapGetter textMapGetter, KafkaTextMapSetter textMapSetter, Tracer tracer, TopologyConfigurationImpl configuration) { - this(delegate, openTelemetry, textMapGetter, textMapSetter, tracer, + this(openTelemetry, textMapGetter, textMapSetter, tracer, configuration.getProcessorPayloadType().getName(), JsonFormat.printer()); } + public TracingDecorator(OpenTelemetry openTelemetry, KafkaTextMapGetter textMapGetter, KafkaTextMapSetter textMapSetter, + Tracer tracer, String applicationName, JsonFormat.Printer jsonPrinter) { + this.openTelemetry = openTelemetry; + this.textMapGetter = textMapGetter; + this.textMapSetter = textMapSetter; + this.tracer = tracer; + this.applicationName = applicationName; + this.jsonPrinter = jsonPrinter; + } + /** * Init just to capture the reference to {@link ProcessorContext}. *

@@ -150,8 +151,8 @@ public TracingDecorator(@Delegate Processor delegate, Open * {@inheritDoc} */ @Override - public void init(final ProcessorContext context) { - delegate.init(context); + public void init(final ProcessorContext context) { + getDelegate().init(context); this.context = context; } @@ -164,7 +165,7 @@ public void init(final ProcessorContext context) { * {@inheritDoc} */ @Override - public void process(Record record) { + public void process(Record record) { SpanBuilder spanBuilder = tracer.spanBuilder(applicationName); final TextMapPropagator propagator = openTelemetry.getPropagators().getTextMapPropagator(); @@ -194,7 +195,7 @@ public void process(Record record) { // the headers in the incoming message so when an outgoing message is produced with the copied // header values it already has the span id from this new child span propagator.inject(Context.current(), record.headers(), textMapSetter); - delegate.process(record); + getDelegate().process(record); span.setStatus(StatusCode.OK); } catch (KafkaException e) { // we got a Kafka exception, we record the exception in the span, log but rethrow the exception @@ -216,7 +217,7 @@ public void process(Record record) { } } - void logInputMessageMetadata(Record record) { + void logInputMessageMetadata(Record record) { if (log.isDebugEnabled()) { Map headers = toMap(record.headers()); LoggedRecord.LoggedRecordBuilder builder = LoggedRecord.builder() diff --git a/impl/src/test/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/CdiRequestContextDecoratorTest.java b/impl/src/test/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/CdiRequestContextDecoratorTest.java index 1f371a8..8c75299 100644 --- a/impl/src/test/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/CdiRequestContextDecoratorTest.java +++ b/impl/src/test/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/CdiRequestContextDecoratorTest.java @@ -41,7 +41,7 @@ @ExtendWith(MockitoExtension.class) class CdiRequestContextDecoratorTest { - CdiRequestContextDecorator decorator; + CdiRequestContextDecorator decorator; @Mock ArcContainer container; @@ -54,8 +54,8 @@ class CdiRequestContextDecoratorTest { @BeforeEach public void setup() { - when(container.requestContext()).thenReturn(requestContext); - decorator = new CdiRequestContextDecorator<>(processor, container); + decorator = new CdiRequestContextDecorator(container); + decorator.setDelegate(processor); } @Test diff --git a/impl/src/test/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/DlqDecoratorTest.java b/impl/src/test/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/DlqDecoratorTest.java index 6c6059a..b0c3074 100644 --- a/impl/src/test/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/DlqDecoratorTest.java +++ b/impl/src/test/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/DlqDecoratorTest.java @@ -62,7 +62,7 @@ public class DlqDecoratorTest { private static final String RECORD_KEY = "key"; private static final String RECORD_VALUE = "value"; - DlqDecorator decorator; + DlqDecorator decorator; DlqProcessorContextDecorator contextDecorator; @@ -86,7 +86,8 @@ public class DlqDecoratorTest { @BeforeEach public void setUp() { - decorator = new DlqDecorator<>(kafkaProcessor, Set.of(FUNCTIONAL_SINK), dlqMetadataHandler, metrics, true); + decorator = new DlqDecorator(Set.of(FUNCTIONAL_SINK), dlqMetadataHandler, metrics, true); + decorator.setDelegate(kafkaProcessor); decorator.init(context); headers = new RecordHeaders(); record = new Record<>(RECORD_KEY, RECORD_VALUE, 0L, headers); @@ -131,7 +132,8 @@ public void shouldForwardKeyValueToAllSinks() { @Test public void shouldDoNothingIfDeactivated() { - decorator = new DlqDecorator<>(kafkaProcessor, Set.of(FUNCTIONAL_SINK), dlqMetadataHandler, metrics, false); + decorator = new DlqDecorator(Set.of(FUNCTIONAL_SINK), dlqMetadataHandler, metrics, false); + decorator.setDelegate(kafkaProcessor); decorator.init(context); decorator.process(record); diff --git a/impl/src/test/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/MetricsDecoratorTest.java b/impl/src/test/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/MetricsDecoratorTest.java index f8ce1a2..bbef50b 100644 --- a/impl/src/test/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/MetricsDecoratorTest.java +++ b/impl/src/test/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/MetricsDecoratorTest.java @@ -48,14 +48,15 @@ public class MetricsDecoratorTest { Ping inputMessage = Ping.newBuilder().setMessage("message").build(); - MetricsDecorator processorDecorator; + MetricsDecorator processorDecorator; @Mock ArcContainer arcContainer; @BeforeEach public void setUp() { - processorDecorator = new MetricsDecorator<>(kafkaProcessor, metrics); + processorDecorator = new MetricsDecorator(metrics); + processorDecorator.setDelegate(kafkaProcessor); } @Test diff --git a/impl/src/test/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/PunctuatorDecorationProcessorDecoratorTest.java b/impl/src/test/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/PunctuatorDecorationProcessorDecoratorTest.java index 5187037..d97b9f2 100644 --- a/impl/src/test/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/PunctuatorDecorationProcessorDecoratorTest.java +++ b/impl/src/test/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/PunctuatorDecorationProcessorDecoratorTest.java @@ -68,9 +68,8 @@ public void process(Record record) { } }; - PunctuatorDecorationProcessorDecorator decorator = new PunctuatorDecorationProcessorDecorator<>( - processor, - decoratedPunctuators); + PunctuatorDecorationProcessorDecorator decorator = new PunctuatorDecorationProcessorDecorator(decoratedPunctuators); + decorator.setDelegate(processor); decorator.init(context); decorator.process(new Record<>("blabla", PingMessage.Ping.newBuilder().setMessage("blabla").build(), 0L, null)); decorator.close(); diff --git a/impl/src/test/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/TracingDecoratorTest.java b/impl/src/test/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/TracingDecoratorTest.java index 930b32f..6e137cc 100644 --- a/impl/src/test/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/TracingDecoratorTest.java +++ b/impl/src/test/java/io/quarkiverse/kafkastreamsprocessor/impl/decorator/processor/TracingDecoratorTest.java @@ -114,7 +114,7 @@ public class TracingDecoratorTest { @RegisterExtension static final OpenTelemetryExtension otel = OpenTelemetryExtension.create(); - TracingDecorator decorator; + TracingDecorator decorator; @Mock JsonFormat.Printer jsonPrinter; @@ -140,8 +140,9 @@ public void setUp() { rootLogger.addHandler(inMemoryLogHandler); rootLogger.setLevel(Level.DEBUG); when(topologyConfiguration.getProcessorPayloadType()).thenReturn((Class) MockType.class); - decorator = new TracingDecorator<>(kafkaProcessor, otel.getOpenTelemetry(), kafkaTextMapGetter, kafkaTextMapSetter, + decorator = new TracingDecorator(otel.getOpenTelemetry(), kafkaTextMapGetter, kafkaTextMapSetter, tracer, topologyConfiguration.getProcessorPayloadType().getName(), jsonPrinter); + decorator.setDelegate(kafkaProcessor); decorator.init(processorContext); } @@ -206,8 +207,9 @@ public void shouldCleanMDCAndScopeInCaseOfException() { .setMessage("blabla") .build(), 0L, headers); - decorator = new TracingDecorator<>(new ThrowExceptionProcessor(), otel.getOpenTelemetry(), kafkaTextMapGetter, + decorator = new TracingDecorator(otel.getOpenTelemetry(), kafkaTextMapGetter, kafkaTextMapSetter, tracer, topologyConfiguration.getProcessorPayloadType().getName(), jsonPrinter); + decorator.setDelegate(new ThrowExceptionProcessor()); decorator.init(processorContext); assertDoesNotThrow(() -> decorator.process(record)); @@ -307,9 +309,9 @@ void shouldLogMetadataEvenIfValueMarshallingToJSONFails() throws Throwable { void shouldLogRawToStringValueIfNotProtobuf() throws Throwable { Processor kafkaProcessor = mock(Processor.class); ProcessorContext processorContext = mock(ProcessorContext.class); - TracingDecorator decorator = new TracingDecorator<>( - kafkaProcessor, GlobalOpenTelemetry.get(), kafkaTextMapGetter, + TracingDecorator decorator = new TracingDecorator(GlobalOpenTelemetry.get(), kafkaTextMapGetter, kafkaTextMapSetter, tracer, topologyConfiguration.getProcessorPayloadType().getName(), jsonPrinter); + decorator.setDelegate(kafkaProcessor); decorator.init(processorContext); RuntimeException exception = new TestException(); @@ -334,9 +336,10 @@ void shouldPropagateOpentelemetryW3CBaggage() { .setPropagators(ContextPropagators.create(TextMapPropagator.composite(W3CTraceContextPropagator.getInstance(), W3CBaggagePropagator.getInstance()))) .build()) { - decorator = new TracingDecorator<>(new LogOpentelemetryBaggageProcessor(), openTelemetryWithBaggageSdk, + decorator = new TracingDecorator(openTelemetryWithBaggageSdk, kafkaTextMapGetter, kafkaTextMapSetter, openTelemetryWithBaggageSdk.getTracer("test"), PROCESSOR_NAME, jsonPrinter); + decorator.setDelegate(new LogOpentelemetryBaggageProcessor()); decorator.init(processorContext); decorator.process(record); diff --git a/integration-tests/custom-serde/pom.xml b/integration-tests/custom-serde/pom.xml index 5b09ec0..196fe9b 100644 --- a/integration-tests/custom-serde/pom.xml +++ b/integration-tests/custom-serde/pom.xml @@ -3,7 +3,7 @@ io.quarkiverse.kafkastreamsprocessor quarkus-kafka-streams-processor-integration-tests - 3.0.0-SNAPSHOT + 4.0.0-SNAPSHOT 4.0.0 diff --git a/integration-tests/custom-serde/src/main/java/io/quarkiverse/kafkastreamsprocessor/sample/customserde/HeaderDecorator.java b/integration-tests/custom-serde/src/main/java/io/quarkiverse/kafkastreamsprocessor/sample/customserde/HeaderDecorator.java index 09889dc..ea5507f 100644 --- a/integration-tests/custom-serde/src/main/java/io/quarkiverse/kafkastreamsprocessor/sample/customserde/HeaderDecorator.java +++ b/integration-tests/custom-serde/src/main/java/io/quarkiverse/kafkastreamsprocessor/sample/customserde/HeaderDecorator.java @@ -22,29 +22,19 @@ import java.nio.charset.StandardCharsets; import jakarta.annotation.Priority; -import jakarta.decorator.Decorator; -import jakarta.decorator.Delegate; -import jakarta.inject.Inject; +import jakarta.enterprise.context.Dependent; import org.apache.kafka.common.header.Header; -import org.apache.kafka.streams.processor.api.Processor; import org.apache.kafka.streams.processor.api.Record; +import io.quarkiverse.kafkastreamsprocessor.api.decorator.processor.AbstractProcessorDecorator; import io.quarkiverse.kafkastreamsprocessor.api.decorator.processor.ProcessorDecoratorPriorities; -@Decorator +@Dependent @Priority(ProcessorDecoratorPriorities.PUNCTUATOR_DECORATION + 2) -public class HeaderDecorator implements Processor { - @lombok.experimental.Delegate(excludes = Excludes.class) - private final Processor delegate; - - @Inject - public HeaderDecorator(@Delegate Processor delegate) { - this.delegate = delegate; - } - +public class HeaderDecorator extends AbstractProcessorDecorator { @Override - public void process(Record record) { + public void process(Record record) { Header header = record.headers().lastHeader("custom-header"); if (header != null) { String value = new String(header.value(), StandardCharsets.UTF_8); @@ -52,11 +42,6 @@ public void process(Record record) { throw new IllegalStateException("Error in header"); } } - delegate.process(record); + getDelegate().process(record); } - - private interface Excludes { - void process(Record record); - } - } diff --git a/integration-tests/json-pojo/pom.xml b/integration-tests/json-pojo/pom.xml index 0f7e5b1..e49c196 100644 --- a/integration-tests/json-pojo/pom.xml +++ b/integration-tests/json-pojo/pom.xml @@ -3,7 +3,7 @@ io.quarkiverse.kafkastreamsprocessor quarkus-kafka-streams-processor-integration-tests - 3.0.0-SNAPSHOT + 4.0.0-SNAPSHOT 4.0.0 diff --git a/integration-tests/kafka-to-rest/pom.xml b/integration-tests/kafka-to-rest/pom.xml index 74eb191..78d1255 100644 --- a/integration-tests/kafka-to-rest/pom.xml +++ b/integration-tests/kafka-to-rest/pom.xml @@ -3,7 +3,7 @@ io.quarkiverse.kafkastreamsprocessor quarkus-kafka-streams-processor-integration-tests - 3.0.0-SNAPSHOT + 4.0.0-SNAPSHOT 4.0.0 diff --git a/integration-tests/multioutput/pom.xml b/integration-tests/multioutput/pom.xml index ec88456..f435e26 100644 --- a/integration-tests/multioutput/pom.xml +++ b/integration-tests/multioutput/pom.xml @@ -3,7 +3,7 @@ io.quarkiverse.kafkastreamsprocessor quarkus-kafka-streams-processor-integration-tests - 3.0.0-SNAPSHOT + 4.0.0-SNAPSHOT 4.0.0 diff --git a/integration-tests/pom.xml b/integration-tests/pom.xml index 819c521..f5edae2 100644 --- a/integration-tests/pom.xml +++ b/integration-tests/pom.xml @@ -4,7 +4,7 @@ io.quarkiverse.kafkastreamsprocessor quarkus-kafka-streams-processor-parent - 3.0.0-SNAPSHOT + 4.0.0-SNAPSHOT quarkus-kafka-streams-processor-integration-tests pom diff --git a/integration-tests/protobuf-binding/pom.xml b/integration-tests/protobuf-binding/pom.xml index f66dc7c..9f48606 100644 --- a/integration-tests/protobuf-binding/pom.xml +++ b/integration-tests/protobuf-binding/pom.xml @@ -3,7 +3,7 @@ io.quarkiverse.kafkastreamsprocessor quarkus-kafka-streams-processor-integration-tests - 3.0.0-SNAPSHOT + 4.0.0-SNAPSHOT 4.0.0 diff --git a/integration-tests/simple/pom.xml b/integration-tests/simple/pom.xml index fc5584f..9cded7b 100644 --- a/integration-tests/simple/pom.xml +++ b/integration-tests/simple/pom.xml @@ -3,7 +3,7 @@ io.quarkiverse.kafkastreamsprocessor quarkus-kafka-streams-processor-integration-tests - 3.0.0-SNAPSHOT + 4.0.0-SNAPSHOT 4.0.0 diff --git a/integration-tests/stateful/pom.xml b/integration-tests/stateful/pom.xml index a3388ed..520baf6 100644 --- a/integration-tests/stateful/pom.xml +++ b/integration-tests/stateful/pom.xml @@ -4,7 +4,7 @@ io.quarkiverse.kafkastreamsprocessor quarkus-kafka-streams-processor-integration-tests - 3.0.0-SNAPSHOT + 4.0.0-SNAPSHOT 4.0.0 diff --git a/pom.xml b/pom.xml index 37c54d4..8bd6740 100644 --- a/pom.xml +++ b/pom.xml @@ -8,7 +8,7 @@ io.quarkiverse.kafkastreamsprocessor quarkus-kafka-streams-processor-parent - 3.0.0-SNAPSHOT + 4.0.0-SNAPSHOT pom bom @@ -96,37 +96,37 @@ LF - - - - - org.codehaus.mojo - license-maven-plugin - 2.4.0 - - - add-license-header - - check-file-header - - process-sources - - - - false - apache_v2 - 2024 - Amadeus s.a.s. - Quarkus Kafka Streams Processor - - **/*.java - - - **/*$$*.java - - - + + + + org.codehaus.mojo + license-maven-plugin + 2.4.0 + + + add-license-header + + check-file-header + + process-sources + + + + false + apache_v2 + 2024 + Amadeus s.a.s. + Quarkus Kafka Streams Processor + + **/*.java + + + **/*$$*.java + + + + diff --git a/runtime/pom.xml b/runtime/pom.xml index 64ed647..7096e7c 100644 --- a/runtime/pom.xml +++ b/runtime/pom.xml @@ -4,7 +4,7 @@ io.quarkiverse.kafkastreamsprocessor quarkus-kafka-streams-processor-parent - 3.0.0-SNAPSHOT + 4.0.0-SNAPSHOT quarkus-kafka-streams-processor diff --git a/spi/pom.xml b/spi/pom.xml index a4233bb..3a657b3 100644 --- a/spi/pom.xml +++ b/spi/pom.xml @@ -4,7 +4,7 @@ io.quarkiverse.kafkastreamsprocessor quarkus-kafka-streams-processor-parent - 3.0.0-SNAPSHOT + 4.0.0-SNAPSHOT quarkus-kafka-streams-processor-spi diff --git a/test-framework/pom.xml b/test-framework/pom.xml index a0d000b..6abde26 100644 --- a/test-framework/pom.xml +++ b/test-framework/pom.xml @@ -4,7 +4,7 @@ io.quarkiverse.kafkastreamsprocessor quarkus-kafka-streams-processor-parent - 3.0.0-SNAPSHOT + 4.0.0-SNAPSHOT quarkus-kafka-streams-processor-test-framework diff --git a/text-map-accessors/pom.xml b/text-map-accessors/pom.xml index a107006..f8b04ad 100644 --- a/text-map-accessors/pom.xml +++ b/text-map-accessors/pom.xml @@ -4,7 +4,7 @@ io.quarkiverse.kafkastreamsprocessor quarkus-kafka-streams-processor-parent - 3.0.0-SNAPSHOT + 4.0.0-SNAPSHOT quarkus-kafka-streams-processor-text-map-accessors