Fix handling of non-deprecated annotations

build.gradle

@@ -1,12 +1,14 @@
 buildscript {
     repositories {
-        maven { url "https://repo.grails.org/grails/core" }
+        mavenCentral()
+        gradlePluginPortal()
     }
     dependencies {
-        classpath "io.micronaut.build.internal:micronaut-gradle-plugins:3.0.3"
+        classpath "io.micronaut.build.internal:micronaut-gradle-plugins:4.0.0.RC5"
     }
 }
 
+
 subprojects { Project subproject ->
     group "io.micronaut.kafka"
 

gradle.properties

@@ -1,5 +1,5 @@
 projectVersion=3.3.1-SNAPSHOT
-micronautDocsVersion=1.0.25
+micronautDocsVersion=2.0.0.RC1
 micronautBuildVersion=1.1.5
 micronautVersion=2.4.2
 groovyVersion=3.0.5

gradle/wrapper/gradle-wrapper.properties

@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-6.8.3-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-7.0-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists

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

@@ -17,7 +17,7 @@
 
 import org.apache.kafka.streams.StreamsBuilder;
 
-import javax.annotation.Nonnull;
+import io.micronaut.core.annotation.NonNull;
 import java.util.Properties;
 
 /**
@@ -44,7 +44,7 @@ public ConfiguredStreamBuilder(Properties configuration) {
      *
      * @return The configuration
      */
-    public @Nonnull Properties getConfiguration() {
+    public @NonNull Properties getConfiguration() {
         return configuration;
     }
 }

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

@@ -16,8 +16,7 @@
 package io.micronaut.configuration.kafka;
 
 import org.apache.kafka.clients.consumer.Consumer;
-
-import javax.annotation.Nonnull;
+import io.micronaut.core.annotation.NonNull;
 
 /**
  * Interface for {@link io.micronaut.configuration.kafka.annotation.KafkaListener} instances to implement
@@ -39,5 +38,5 @@ public interface ConsumerAware<K, V> {
      *
      * @param consumer The consumer
      */
-    void setKafkaConsumer(@Nonnull Consumer<K, V> consumer);
+    void setKafkaConsumer(@NonNull Consumer<K, V> consumer);
 }

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

@@ -18,7 +18,7 @@
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.common.TopicPartition;
 
-import javax.annotation.Nonnull;
+import io.micronaut.core.annotation.NonNull;
 import java.util.Set;
 
 /**
@@ -39,8 +39,8 @@ public interface ConsumerRegistry {
      * @return The consumer
      * @throws IllegalArgumentException If no consumer exists for the given ID
      */
-    @Nonnull
-    <K, V> Consumer<K, V> getConsumer(@Nonnull String id);
+    @NonNull
+    <K, V> Consumer<K, V> getConsumer(@NonNull String id);
 
     /**
      * Returns a managed Consumer's subscriptions.
@@ -49,8 +49,8 @@ public interface ConsumerRegistry {
      * @return The consumer subscription
      * @throws IllegalArgumentException If no consumer exists for the given ID
      */
-    @Nonnull
-    Set<String> getConsumerSubscription(@Nonnull String id);
+    @NonNull
+    Set<String> getConsumerSubscription(@NonNull String id);
 
     /**
      * Returns a managed Consumer's assignment info.
@@ -59,23 +59,23 @@ public interface ConsumerRegistry {
      * @return The consumer assignment
      * @throws IllegalArgumentException If no consumer exists for the given ID
      */
-    @Nonnull
-    Set<TopicPartition> getConsumerAssignment(@Nonnull String id);
+    @NonNull
+    Set<TopicPartition> getConsumerAssignment(@NonNull String id);
 
     /**
      * The IDs of the available consumers.
      *
      * @return The consumers
      */
-    @Nonnull Set<String> getConsumerIds();
+    @NonNull Set<String> getConsumerIds();
 
     /**
      * Is the consumer with the given ID paused.
      *
      * @param id True it is paused
      * @return True if it is paused
      */
-    boolean isPaused(@Nonnull String id);
+    boolean isPaused(@NonNull String id);
 
     /**
      * Pause the consumer for the given ID. Note that this method will request that the consumer is paused, however
@@ -84,7 +84,7 @@ public interface ConsumerRegistry {
      *
      * @param id The id of the consumer
      */
-    void pause(@Nonnull String id);
+    void pause(@NonNull String id);
 
 
     /**
@@ -94,5 +94,5 @@ public interface ConsumerRegistry {
      *
      * @param id The id of the consumer
      */
-    void resume(@Nonnull String id);
+    void resume(@NonNull String id);
 }

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

@@ -18,7 +18,7 @@
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 
-import javax.annotation.Nonnull;
+import io.micronaut.core.annotation.NonNull;
 
 /**
  * Interface for {@link io.micronaut.configuration.kafka.annotation.KafkaListener} instances to implement
@@ -38,10 +38,10 @@ public interface KafkaConsumerAware<K, V> extends ConsumerAware<K, V> {
      *
      * @param consumer The consumer
      */
-    void setKafkaConsumer(@Nonnull KafkaConsumer<K, V> consumer);
+    void setKafkaConsumer(@NonNull KafkaConsumer<K, V> consumer);
 
     @Override
-    default void setKafkaConsumer(@Nonnull Consumer<K, V> consumer) {
+    default void setKafkaConsumer(@NonNull Consumer<K, V> consumer) {
         if (consumer instanceof KafkaConsumer) {
             setKafkaConsumer((KafkaConsumer<K, V>) consumer);
         }

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

@@ -18,7 +18,7 @@
 import io.micronaut.core.type.Argument;
 import org.apache.kafka.clients.producer.KafkaProducer;
 
-import javax.annotation.Nonnull;
+import io.micronaut.core.annotation.NonNull;
 
 /**
  * A registry of managed {@link KafkaProducer} instances key by id and type.
@@ -41,5 +41,5 @@ public interface KafkaProducerRegistry extends ProducerRegistry {
      * @return The producer
      */
     @Override
-    @Nonnull <K, V> KafkaProducer<K, V> getProducer(String id, Argument<K> keyType, Argument<V> valueType);
+    @NonNull <K, V> KafkaProducer<K, V> getProducer(String id, Argument<K> keyType, Argument<V> valueType);
 }

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

@@ -18,7 +18,7 @@
 import io.micronaut.core.type.Argument;
 import org.apache.kafka.clients.producer.Producer;
 
-import javax.annotation.Nonnull;
+import io.micronaut.core.annotation.NonNull;
 
 /**
  * A registry of managed {@link Producer} instances key by id and type.
@@ -37,6 +37,6 @@ public interface ProducerRegistry {
      * @param <V> The value generic type
      * @return The producer
      */
-    @Nonnull
+    @NonNull
     <K, V> Producer<K, V> getProducer(String id, Argument<K> keyType, Argument<V> valueType);
 }

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

@@ -33,6 +33,7 @@
  * @since 1.0
  */
 @Singleton
+@Deprecated
 public class KafkaHeaderBinder<T> implements AnnotatedConsumerRecordBinder<Header, T> {
 
     @Override

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

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2017-2021 original authors
+ *
+ * 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
+ *
+ * https://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.
+ */
+package io.micronaut.configuration.kafka.bind;
+
+import io.micronaut.core.annotation.AnnotationMetadata;
+import io.micronaut.core.convert.ArgumentConversionContext;
+import io.micronaut.core.convert.ConversionService;
+import io.micronaut.messaging.annotation.Header;
+import io.micronaut.messaging.annotation.MessageHeader;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.header.Headers;
+
+import javax.inject.Singleton;
+import java.util.Optional;
+
+@Singleton
+public class KafkaMessageHeaderBinder<T> implements AnnotatedConsumerRecordBinder<MessageHeader, T> {
+
+    @Override
+    public Class<MessageHeader> annotationType() {
+        return MessageHeader.class;
+    }
+
+    @Override
+    public BindingResult<T> bind(ArgumentConversionContext<T> context, ConsumerRecord<?, ?> source) {
+        Headers headers = source.headers();
+        AnnotationMetadata annotationMetadata = context.getAnnotationMetadata();
+
+        // use deprecated versions as that is what is stored in metadata
+        String name = annotationMetadata.stringValue(Header.class, "name")
+                .orElseGet(() -> annotationMetadata.stringValue(Header.class)
+                        .orElse(context.getArgument().getName()));
+        Iterable<org.apache.kafka.common.header.Header> value = headers.headers(name);
+
+        if (value.iterator().hasNext()) {
+            Optional<T> converted = ConversionService.SHARED.convert(value, context);
+            return () -> converted;
+        } else if (context.getArgument().getType() == Optional.class) {
+            //noinspection unchecked
+            return () -> (Optional<T>) Optional.of(Optional.empty());
+        } else {
+            //noinspection unchecked
+            return BindingResult.EMPTY;
+        }
+    }
+}

kafka/src/main/java/io/micronaut/configuration/kafka/config/AbstractKafkaConfiguration.java

@@ -17,7 +17,7 @@
 
 import io.micronaut.core.util.Toggleable;
 
-import javax.annotation.Nonnull;
+import io.micronaut.core.annotation.NonNull;
 import java.util.Properties;
 
 /**
@@ -71,7 +71,7 @@ protected AbstractKafkaConfiguration(Properties config) {
     /**
      * @return The Kafka configuration
      */
-    public @Nonnull Properties getConfig() {
+    public @NonNull Properties getConfig() {
         if (config != null) {
             return config;
         }

kafka/src/main/java/io/micronaut/configuration/kafka/config/AbstractKafkaConsumerConfiguration.java

@@ -17,7 +17,7 @@
 
 import org.apache.kafka.common.serialization.Deserializer;
 
-import javax.annotation.Nullable;
+import io.micronaut.core.annotation.Nullable;
 import java.util.Optional;
 import java.util.Properties;
 

kafka/src/main/java/io/micronaut/configuration/kafka/config/AbstractKafkaProducerConfiguration.java

@@ -17,7 +17,7 @@
 
 import org.apache.kafka.common.serialization.Serializer;
 
-import javax.annotation.Nullable;
+import io.micronaut.core.annotation.Nullable;
 import java.util.Optional;
 import java.util.Properties;
 

kafka/src/main/java/io/micronaut/configuration/kafka/embedded/KafkaEmbeddedConfiguration.java

@@ -19,7 +19,7 @@
 import io.micronaut.context.annotation.ConfigurationProperties;
 import io.micronaut.core.util.Toggleable;
 
-import javax.annotation.Nonnull;
+import io.micronaut.core.annotation.NonNull;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Properties;
@@ -65,7 +65,7 @@ public void setEnabled(boolean enabled) {
     /**
      * @return The broker properties.
      */
-    public @Nonnull Properties getProperties() {
+    public @NonNull Properties getProperties() {
         return properties;
     }
 

kafka/src/main/java/io/micronaut/configuration/kafka/exceptions/DefaultKafkaListenerExceptionHandler.java

@@ -23,7 +23,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.annotation.Nonnull;
+import io.micronaut.core.annotation.NonNull;
 import javax.inject.Singleton;
 import java.util.Optional;
 import java.util.regex.Matcher;
@@ -82,9 +82,9 @@ public void setSkipRecordOnDeserializationFailure(boolean skipRecordOnDeserializ
      * @param kafkaConsumer The kafka consumer
      */
     protected void seekPastDeserializationError(
-            @Nonnull SerializationException cause,
-            @Nonnull Object consumerBean,
-            @Nonnull Consumer<?, ?> kafkaConsumer) {
+            @NonNull SerializationException cause,
+            @NonNull Object consumerBean,
+            @NonNull Consumer<?, ?> kafkaConsumer) {
         try {
             final String message = cause.getMessage();
             final Matcher matcher = SERIALIZATION_EXCEPTION_MESSAGE_PATTERN.matcher(message);

kafka/src/main/java/io/micronaut/configuration/kafka/intercept/KafkaClientIntroductionAdvice.java

@@ -56,7 +56,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.annotation.Nullable;
+import io.micronaut.core.annotation.Nullable;
 import javax.annotation.PreDestroy;
 import javax.inject.Singleton;
 import java.nio.charset.StandardCharsets;