Add health-enabled attribute to AMQP connector

Fixes #2990

Author: ozangunalp

smallrye-reactive-messaging-amqp/src/main/java/io/smallrye/reactive/messaging/amqp/AmqpConnector.java

@@ -56,6 +56,7 @@
 @ConnectorAttribute(name = "client-options-name", direction = INCOMING_AND_OUTGOING, description = "The name of the AMQP Client Option bean used to customize the AMQP client configuration", type = "string", alias = "amqp-client-options-name")
 @ConnectorAttribute(name = "client-ssl-context-name", direction = INCOMING_AND_OUTGOING, description = "The name of an SSLContext bean to use for connecting to AMQP when SSL is used", type = "string", alias = "amqp-client-ssl-context-name", hiddenFromDocumentation = true)
 @ConnectorAttribute(name = "tracing-enabled", direction = INCOMING_AND_OUTGOING, description = "Whether tracing is enabled (default) or disabled", type = "boolean", defaultValue = "true")
+@ConnectorAttribute(name = "health-enabled", type = "boolean", direction = INCOMING_AND_OUTGOING, description = "Whether health reporting is enabled (default) or disabled", defaultValue = "true")
 @ConnectorAttribute(name = "health-timeout", direction = INCOMING_AND_OUTGOING, description = "The max number of seconds to wait to determine if the connection with the broker is still established for the readiness check. After that threshold, the check is considered as failed.", type = "int", defaultValue = "3")
 @ConnectorAttribute(name = "cloud-events", type = "boolean", direction = INCOMING_AND_OUTGOING, description = "Enables (default) or disables the Cloud Event support. If enabled on an _incoming_ channel, the connector analyzes the incoming records and try to create Cloud Event metadata. If enabled on an _outgoing_, the connector sends the outgoing messages as Cloud Event if the message includes Cloud Event Metadata.", defaultValue = "true")
 @ConnectorAttribute(name = "capabilities", type = "string", direction = INCOMING_AND_OUTGOING, description = " A comma-separated list of capabilities proposed by the sender or receiver client.")
@@ -182,20 +183,24 @@ public List<AmqpClient> getClients() {
     public HealthReport getReadiness() {
         HealthReport.HealthReportBuilder builder = HealthReport.builder();
         for (Map.Entry<String, IncomingAmqpChannel> holder : incomingChannels.entrySet()) {
-            try {
-                builder.add(holder.getKey(), holder.getValue().isConnected().await()
-                        .atMost(Duration.ofSeconds(holder.getValue().getHealthTimeout())));
-            } catch (Exception e) {
-                builder.add(holder.getKey(), false, e.getMessage());
+            if (holder.getValue().isHealthEnabled()) {
+                try {
+                    builder.add(holder.getKey(), holder.getValue().isConnected().await()
+                            .atMost(Duration.ofSeconds(holder.getValue().getHealthTimeout())));
+                } catch (Exception e) {
+                    builder.add(holder.getKey(), false, e.getMessage());
+                }
             }
         }
 
         for (Map.Entry<String, OutgoingAmqpChannel> sender : outgoingChannels.entrySet()) {
-            try {
-                builder.add(sender.getKey(), sender.getValue().isConnected().await()
-                        .atMost(Duration.ofSeconds(sender.getValue().getHealthTimeout())));
-            } catch (Exception e) {
-                builder.add(sender.getKey(), false, e.getMessage());
+            if (sender.getValue().isHealthEnabled()) {
+                try {
+                    builder.add(sender.getKey(), sender.getValue().isConnected().await()
+                            .atMost(Duration.ofSeconds(sender.getValue().getHealthTimeout())));
+                } catch (Exception e) {
+                    builder.add(sender.getKey(), false, e.getMessage());
+                }
             }
         }
 
@@ -212,10 +217,14 @@ public HealthReport getReadiness() {
     public HealthReport getLiveness() {
         HealthReport.HealthReportBuilder builder = HealthReport.builder();
         for (Map.Entry<String, IncomingAmqpChannel> entry : incomingChannels.entrySet()) {
-            builder.add(entry.getKey(), entry.getValue().isOpen());
+            if (entry.getValue().isHealthEnabled()) {
+                builder.add(entry.getKey(), entry.getValue().isOpen());
+            }
         }
         for (Map.Entry<String, OutgoingAmqpChannel> entry : outgoingChannels.entrySet()) {
-            builder.add(entry.getKey(), entry.getValue().isOpen());
+            if (entry.getValue().isHealthEnabled()) {
+                builder.add(entry.getKey(), entry.getValue().isOpen());
+            }
         }
         return builder.build();
     }

smallrye-reactive-messaging-amqp/src/main/java/io/smallrye/reactive/messaging/amqp/IncomingAmqpChannel.java

@@ -39,11 +39,13 @@ public class IncomingAmqpChannel {
     private final AtomicBoolean opened;
     private final BiConsumer<String, Throwable> reportFailure;
     private final ConnectionHolder holder;
+    private final boolean healthEnabled;
 
     public IncomingAmqpChannel(AmqpConnectorIncomingConfiguration ic, AmqpClient client, Vertx vertx,
             Instance<OpenTelemetry> openTelemetryInstance, BiConsumer<String, Throwable> reportFailure) {
         this.reportFailure = reportFailure;
         this.opened = new AtomicBoolean(false);
+        this.healthEnabled = ic.getHealthEnabled();
 
         String channel = ic.getChannel();
         String address = ic.getAddress().orElse(channel);
@@ -160,6 +162,10 @@ public long getHealthTimeout() {
         return holder.getHealthTimeout();
     }
 
+    public boolean isHealthEnabled() {
+        return healthEnabled;
+    }
+
     public void close() {
         opened.set(false);
     }

smallrye-reactive-messaging-amqp/src/main/java/io/smallrye/reactive/messaging/amqp/OutgoingAmqpChannel.java

@@ -26,12 +26,14 @@ public class OutgoingAmqpChannel {
     private final AtomicBoolean opened;
     private final Flow.Subscriber<Message<?>> subscriber;
     private final AmqpCreditBasedSender processor;
+    private final boolean healthEnabled;
 
     public OutgoingAmqpChannel(AmqpConnectorOutgoingConfiguration oc, AmqpClient client, Vertx vertx,
             Instance<OpenTelemetry> openTelemetryInstance, BiConsumer<String, Throwable> reportFailure) {
         String configuredAddress = oc.getAddress().orElseGet(oc::getChannel);
 
         opened = new AtomicBoolean(false);
+        healthEnabled = oc.getHealthEnabled();
 
         AtomicReference<AmqpSender> sender = new AtomicReference<>();
         String link = oc.getLinkName().orElseGet(oc::getChannel);
@@ -131,6 +133,10 @@ public long getHealthTimeout() {
         return processor.getHealthTimeout();
     }
 
+    public boolean isHealthEnabled() {
+        return healthEnabled;
+    }
+
     public void close() {
         processor.cancel();
         opened.set(false);