Merge pull request #49861 from RoMiRoSSaN/issue/48641

Correct setting kafka graceful-shutdown property for dev and test profiles

Author: ozangunalp

extensions/smallrye-reactive-messaging-kafka/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/SmallRyeReactiveMessagingKafkaProcessor.java

@@ -42,6 +42,7 @@
 import io.quarkus.deployment.logging.LogCleanupFilterBuildItem;
 import io.quarkus.hibernate.orm.deployment.spi.AdditionalJpaModelBuildItem;
 import io.quarkus.smallrye.reactivemessaging.deployment.ReactiveMessagingDotNames;
+import io.quarkus.smallrye.reactivemessaging.deployment.items.ChannelDirection;
 import io.quarkus.smallrye.reactivemessaging.deployment.items.ConnectorManagedChannelBuildItem;
 import io.quarkus.smallrye.reactivemessaging.kafka.DatabindProcessingStateCodec;
 import io.quarkus.smallrye.reactivemessaging.kafka.HibernateOrmStateStore;
@@ -201,27 +202,27 @@ public void defaultChannelConfiguration(
 
         if (launchMode.getLaunchMode().isDevOrTest()) {
             if (!buildTimeConfig.enableGracefulShutdownInDevAndTestMode()) {
-                List<AnnotationInstance> incomings = discoveryState.findRepeatableAnnotationsOnMethods(DotNames.INCOMING);
-                List<AnnotationInstance> outgoings = discoveryState.findRepeatableAnnotationsOnMethods(DotNames.OUTGOING);
-                List<AnnotationInstance> channels = discoveryState.findAnnotationsOnInjectionPoints(DotNames.CHANNEL);
-                List<AnnotationInstance> annotations = new ArrayList<>();
-                annotations.addAll(incomings);
-                annotations.addAll(outgoings);
-                annotations.addAll(channels);
-                for (AnnotationInstance annotation : annotations) {
-                    String channelName = annotation.value().asString();
-                    if (!discoveryState.isKafkaConnector(channelsManagedByConnectors, true, channelName)) {
-                        continue;
-                    }
-                    String key = getChannelPropertyKey(channelName, "graceful-shutdown", true);
-                    discoveryState.ifNotYetConfigured(key, () -> {
-                        defaultConfigProducer.produce(new RunTimeConfigurationDefaultBuildItem(key, "false"));
-                    });
-                }
+                disableGracefulShutdown(channelsManagedByConnectors, defaultConfigProducer, discoveryState);
             }
         }
     }
 
+    void disableGracefulShutdown(List<ConnectorManagedChannelBuildItem> channelsManagedByConnectors,
+            BuildProducer<RunTimeConfigurationDefaultBuildItem> defaultConfigProducer,
+            DefaultSerdeDiscoveryState discoveryState) {
+        for (ConnectorManagedChannelBuildItem managed : channelsManagedByConnectors) {
+            String channelName = managed.getName();
+            boolean incoming = managed.getDirection() == ChannelDirection.INCOMING;
+            if (!discoveryState.isKafkaConnector(channelsManagedByConnectors, incoming, channelName)) {
+                continue;
+            }
+            String key = getChannelPropertyKey(channelName, incoming ? "graceful-shutdown" : "close-timeout", incoming);
+            discoveryState.ifNotYetConfigured(key, () -> {
+                defaultConfigProducer.produce(new RunTimeConfigurationDefaultBuildItem(key, incoming ? "false" : "0"));
+            });
+        }
+    }
+
     // visible for testing
     void discoverDefaultSerdeConfig(DefaultSerdeDiscoveryState discovery,
             List<ConnectorManagedChannelBuildItem> channelsManagedByConnectors,