Upgrade to Micronaut 4 snapshot

.github/workflows/central-sync.yml

@@ -22,8 +22,8 @@ jobs:
       - name: Set up JDK
         uses: actions/setup-java@v3
         with:
-          distribution: 'temurin'
-          java-version: '11'
+          distribution: 'adopt'
+          java-version: '17'
       - name: Publish to Sonatype OSSRH
         env:
           SONATYPE_USERNAME: ${{ secrets.SONATYPE_USERNAME }}

.github/workflows/graalvm.yml

@@ -19,7 +19,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        java: ['11', '17']
+        java: ['17']
         graalvm: ['latest', 'dev']
     steps:
        # https://github.com/actions/virtual-environments/issues/709

.github/workflows/gradle.yml

@@ -19,7 +19,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        java: ['8', '11', '17']
+        java: ['17']
     steps:
        # https://github.com/actions/virtual-environments/issues/709
       - name: Free disk space
@@ -72,7 +72,7 @@ jobs:
           name: binary-compatibility-reports
           path: "**/build/reports/binary-compatibility-*.html"
       - name: Publish to Sonatype Snapshots
-        if: success() && github.event_name == 'push' && matrix.java == '11'
+        if: success() && github.event_name == 'push' && matrix.java == '17'
         env:
           SONATYPE_USERNAME: ${{ secrets.SONATYPE_USERNAME }}
           SONATYPE_PASSWORD: ${{ secrets.SONATYPE_PASSWORD }}
@@ -88,7 +88,7 @@ jobs:
           if_true: "micronaut-projects/micronaut-docs"
           if_false: ${{ github.repository }}
       - name: Publish to Github Pages
-        if: success() && github.event_name == 'push' && matrix.java == '11'
+        if: success() && github.event_name == 'push' && matrix.java == '17'
         uses: micronaut-projects/github-pages-deploy-action@master
         env:
           TARGET_REPOSITORY: ${{ steps.docs_target.outputs.value }}

.github/workflows/publish-snapshot.yml

@@ -20,8 +20,8 @@ jobs:
       - name: Set up JDK
         uses: actions/setup-java@v3
         with:
-          distribution: 'temurin'
-          java-version: '11'
+          distribution: 'adopt'
+          java-version: '17'
       - name: Publish to Sonatype Snapshots
         if: success()
         env:

.github/workflows/release.yml

@@ -21,8 +21,8 @@ jobs:
       - name: Set up JDK
         uses: actions/setup-java@v3
         with:
-          distribution: 'temurin'
-          java-version: '11'
+          distribution: 'adopt'
+          java-version: '17'
       - name: Set the current release version
         id: release_version
         run: echo ::set-output name=release_version::${GITHUB_REF:11}

.github/workflows/sonarqube.yml

@@ -37,8 +37,8 @@ jobs:
       - name: Set up JDK
         uses: actions/setup-java@v3
         with:
-          distribution: 'temurin'
-          java-version: 11
+          distribution: 'adopt'
+          java-version: 17
       - name: Optional setup step
         env:
           GRADLE_ENTERPRISE_ACCESS_KEY: ${{ secrets.GRADLE_ENTERPRISE_ACCESS_KEY }}

build.gradle

@@ -1,5 +1,9 @@
 plugins {
     id 'io.micronaut.build.internal.docs'
-    id 'io.micronaut.build.internal.dependency-updates'
     id 'io.micronaut.build.internal.quality-reporting'
 }
+
+repositories {
+    mavenCentral()
+    maven { url "https://s01.oss.sonatype.org/content/repositories/snapshots/" }
+}

buildSrc/src/main/groovy/io.micronaut.build.internal.kafka-base.gradle

@@ -0,0 +1,11 @@
+repositories {
+    mavenCentral()
+    maven { url "https://s01.oss.sonatype.org/content/repositories/snapshots/" }
+}
+
+configurations.all {
+    resolutionStrategy.dependencySubstitution {
+        substitute(module("org.codehaus.groovy:groovy"))
+                .using(module("org.apache.groovy:groovy:4.0.6"))
+    }
+}

buildSrc/src/main/groovy/io.micronaut.build.internal.kafka-module.gradle

@@ -1,9 +1,11 @@
 plugins {
     id 'io.micronaut.build.internal.module'
+    id 'io.micronaut.build.internal.kafka-base'
 }
 
 dependencies {
     api mn.micronaut.messaging
+    api mn.micronaut.json.core
 
     compileOnly mn.micronaut.management
 
@@ -19,6 +21,7 @@ dependencies {
     testRuntimeOnly mn.micronaut.reactor
     testRuntimeOnly mn.micronaut.http.server.netty
     testRuntimeOnly mn.micronaut.management
+    testRuntimeOnly mn.snakeyaml
 }
 
 configurations.all {

buildSrc/src/main/groovy/io.micronaut.build.internal.kafka-tests.gradle

@@ -0,0 +1,3 @@
+plugins {
+    id 'io.micronaut.build.internal.kafka-base'
+}

gradle.properties

@@ -1,4 +1,4 @@
-projectVersion=4.5.1-SNAPSHOT
+projectVersion=5.0.0-SNAPSHOT
 projectGroup=io.micronaut.kafka
 
 title=Micronaut Kafka
@@ -7,7 +7,7 @@ projectUrl=https://micronaut.io
 githubSlug=micronaut-projects/micronaut-kafka
 developers=Graeme Rocher
 
-githubCoreBranch=3.8.x
+githubCoreBranch=4.0.x
 bomProperty=micronautKafkaVersion
 
 testskafka=kafka/src/test/groovy/io/micronaut/configuration/kafka/docs

gradle/libs.versions.toml

@@ -1,12 +1,12 @@
 [versions]
-micronaut = "3.7.3"
+micronaut = "4.0.0-SNAPSHOT"
 micronaut-docs = "2.0.0"
 
 # Required to keep catalog compatibility with 3.4.x.  Can be removed for 4.0.0
 managed-kafka-compat = "2.8.2"
 managed-kafka = '2.8.2'
 
-groovy = "3.0.11"
+groovy = "4.0.6"
 
 opentracing-kafka-client = '0.1.15'
 opentracing-mock = '0.33.0'

kafka-streams/build.gradle

@@ -7,4 +7,5 @@ dependencies {
     api libs.managed.kafka.streams
     compileOnly mn.micronaut.micrometer.core
     testImplementation mn.micronaut.http.client
+    testImplementation mn.micronaut.serde.jackson
 }

kafka-streams/src/main/java/io/micronaut/configuration/kafka/streams/KafkaStreamsFactory.java

@@ -17,8 +17,14 @@
 
 import io.micronaut.configuration.kafka.streams.event.AfterKafkaStreamsStart;
 import io.micronaut.configuration.kafka.streams.event.BeforeKafkaStreamStart;
-import io.micronaut.context.annotation.*;
+import io.micronaut.context.annotation.Context;
+import io.micronaut.context.annotation.EachBean;
+import io.micronaut.context.annotation.Factory;
+import io.micronaut.context.annotation.Parameter;
+import io.micronaut.context.annotation.Secondary;
 import io.micronaut.context.event.ApplicationEventPublisher;
+import jakarta.annotation.PreDestroy;
+import jakarta.inject.Singleton;
 import org.apache.kafka.streams.KafkaClientSupplier;
 import org.apache.kafka.streams.KafkaStreams;
 import org.apache.kafka.streams.Topology;
@@ -27,8 +33,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.annotation.PreDestroy;
-import jakarta.inject.Singleton;
 import java.io.Closeable;
 import java.time.Duration;
 import java.util.Map;

kafka-streams/src/main/java/io/micronaut/configuration/kafka/streams/metrics/KafkaStreamsMetricsReporter.java

@@ -16,8 +16,8 @@
 package io.micronaut.configuration.kafka.streams.metrics;
 
 import io.micronaut.configuration.kafka.metrics.AbstractKafkaMetricsReporter;
+import jakarta.annotation.PreDestroy;
 
-import javax.annotation.PreDestroy;
 
 /**
  * Kafka streams specific metrics reporter which prefixes all metrics with kafka-streams.

kafka/build.gradle

@@ -19,6 +19,7 @@ dependencies {
     compileOnly mn.micronaut.graal
     compileOnly mn.graal
 
+    testImplementation mn.micronaut.serde.jackson
     testImplementation mn.micronaut.rxjava2
     testImplementation mn.micronaut.http.client
     testImplementation libs.opentracing.mock

kafka/src/main/java/io/micronaut/configuration/kafka/KafkaHeaders.java

@@ -36,15 +36,29 @@
 public class KafkaHeaders implements MessageHeaders {
 
     private final Headers headers;
+    private final ConversionService conversionService;
 
     /**
      * Constructs a new instance for the given headers.
      *
      * @param headers The kafka headers
      */
+    @Deprecated
     public KafkaHeaders(Headers headers) {
+        this(headers, ConversionService.SHARED);
+    }
+
+    /**
+     * Constructs a new instance for the given headers.
+     *
+     * @param headers The kafka headers
+     * @param conversionService The conversion service
+     */
+    public KafkaHeaders(Headers headers, ConversionService conversionService) {
         Objects.requireNonNull(headers, "Argument [headers] cannot be null");
+        Objects.requireNonNull(conversionService, "Argument [conversionService] cannot be null");
         this.headers = headers;
+        this.conversionService = conversionService;
     }
 
     @Override
@@ -87,7 +101,7 @@ public Collection<List<String>> values() {
     public <T> Optional<T> get(CharSequence name, ArgumentConversionContext<T> conversionContext) {
         String v = get(name);
         if (v != null) {
-            return ConversionService.SHARED.convert(v, conversionContext);
+            return conversionService.convert(v, conversionContext);
         }
         return Optional.empty();
     }

kafka/src/main/java/io/micronaut/configuration/kafka/KafkaProducerFactory.java

@@ -24,16 +24,6 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Supplier;
 
-import javax.annotation.PreDestroy;
-
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.producer.KafkaProducer;
-import org.apache.kafka.clients.producer.Producer;
-import org.apache.kafka.clients.producer.ProducerConfig;
-import org.apache.kafka.common.serialization.Serializer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import io.micronaut.configuration.kafka.annotation.KafkaClient;
 import io.micronaut.configuration.kafka.config.AbstractKafkaProducerConfiguration;
 import io.micronaut.configuration.kafka.config.DefaultKafkaProducerConfiguration;
@@ -53,6 +43,14 @@
 import io.micronaut.inject.FieldInjectionPoint;
 import io.micronaut.inject.InjectionPoint;
 import io.micronaut.inject.qualifiers.Qualifiers;
+import jakarta.annotation.PreDestroy;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.Serializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * A registry class for Kafka {@link org.apache.kafka.clients.producer.Producer} instances.

kafka/src/main/java/io/micronaut/configuration/kafka/annotation/KafkaClient.java

@@ -18,7 +18,6 @@
 import io.micronaut.aop.Introduction;
 import io.micronaut.context.annotation.AliasFor;
 import io.micronaut.context.annotation.Property;
-import io.micronaut.retry.annotation.Recoverable;
 import jakarta.inject.Singleton;
 
 import java.lang.annotation.Documented;
@@ -37,7 +36,6 @@
 @Documented
 @Retention(RUNTIME)
 @Introduction
-@Recoverable
 @Singleton
 public @interface KafkaClient {
 

kafka/src/main/java/io/micronaut/configuration/kafka/bind/ConsumerRecordBinderRegistry.java

@@ -18,14 +18,17 @@
 import io.micronaut.core.bind.ArgumentBinder;
 import io.micronaut.core.bind.ArgumentBinderRegistry;
 import io.micronaut.core.bind.annotation.Bindable;
+import io.micronaut.core.convert.ConversionService;
 import io.micronaut.core.type.Argument;
 import io.micronaut.core.util.ArrayUtils;
+import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 
 import java.lang.annotation.Annotation;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 
 /**
@@ -45,8 +48,21 @@ public class ConsumerRecordBinderRegistry implements ArgumentBinderRegistry<Cons
      * Creates the registry for the given binders.
      *
      * @param binders The binders
+     * @deprecated Use conversion service constructor instead.
      */
+    @Deprecated
     public ConsumerRecordBinderRegistry(ConsumerRecordBinder<?>... binders) {
+        this(ConversionService.SHARED, binders);
+    }
+
+    /**
+     * Creates the registry for the given binders.
+     *
+     * @param conversionService The conversion service
+     * @param binders The binders
+     */
+    @Inject
+    public ConsumerRecordBinderRegistry(ConversionService conversionService, ConsumerRecordBinder<?>... binders) {
         if (ArrayUtils.isNotEmpty(binders)) {
             for (ConsumerRecordBinder<?> binder : binders) {
                 if (binder instanceof AnnotatedConsumerRecordBinder) {
@@ -77,11 +93,7 @@ public ConsumerRecordBinderRegistry(ConsumerRecordBinder<?>... binders) {
         } else {
             @SuppressWarnings("unchecked")
             ConsumerRecordBinder<T> binder = (ConsumerRecordBinder<T>) byType.get(argument.typeHashCode());
-            if (binder != null) {
-                return Optional.of(binder);
-            } else {
-                return Optional.of(new KafkaDefaultBinder<>());
-            }
+            return Optional.of(Objects.requireNonNullElseGet(binder, KafkaDefaultBinder::new));
         }
     }
 }

kafka/src/main/java/io/micronaut/configuration/kafka/bind/KafkaHeadersBinder.java

@@ -17,6 +17,7 @@
 
 import io.micronaut.configuration.kafka.KafkaHeaders;
 import io.micronaut.core.convert.ArgumentConversionContext;
+import io.micronaut.core.convert.ConversionService;
 import io.micronaut.core.type.Argument;
 import io.micronaut.messaging.MessageHeaders;
 import jakarta.inject.Singleton;
@@ -35,15 +36,20 @@ public class KafkaHeadersBinder implements TypedConsumerRecordBinder<MessageHead
 
     public static final Argument<MessageHeaders> TYPE = Argument.of(MessageHeaders.class);
 
+    private final ConversionService conversionService;
+
+    public KafkaHeadersBinder(ConversionService conversionService) {
+        this.conversionService = conversionService;
+    }
+
     @Override
     public Argument<MessageHeaders> argumentType() {
         return TYPE;
     }
 
     @Override
     public BindingResult<MessageHeaders> bind(ArgumentConversionContext<MessageHeaders> context, ConsumerRecord<?, ?> source) {
-
-        KafkaHeaders kafkaHeaders = new KafkaHeaders(source.headers());
+        KafkaHeaders kafkaHeaders = new KafkaHeaders(source.headers(), conversionService);
         return () -> Optional.of(kafkaHeaders);
     }
 }