[CAMEL-22401] Add retry strategy for Couchbase GET operation + consumerRetryPause (#19064)

Author: llowinge

catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/couchbase.json

@@ -41,6 +41,8 @@ public boolean configure(CamelContext camelContext, Object obj, String name, Obj
         case "connectTimeout": target.setConnectTimeout(property(camelContext, java.time.Duration.class, value).toMillis()); return true;
         case "consumerprocessedstrategy":
         case "consumerProcessedStrategy": target.setConsumerProcessedStrategy(property(camelContext, java.lang.String.class, value)); return true;
+        case "consumerretrypause":
+        case "consumerRetryPause": target.setConsumerRetryPause(property(camelContext, int.class, value)); return true;
         case "delay": target.setDelay(property(camelContext, long.class, value)); return true;
         case "descending": target.setDescending(property(camelContext, boolean.class, value)); return true;
         case "designdocumentname":
@@ -125,6 +127,8 @@ public Class<?> getOptionType(String name, boolean ignoreCase) {
         case "connectTimeout": return long.class;
         case "consumerprocessedstrategy":
         case "consumerProcessedStrategy": return java.lang.String.class;
+        case "consumerretrypause":
+        case "consumerRetryPause": return int.class;
         case "delay": return long.class;
         case "descending": return boolean.class;
         case "designdocumentname":
@@ -210,6 +214,8 @@ public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
         case "connectTimeout": return target.getConnectTimeout();
         case "consumerprocessedstrategy":
         case "consumerProcessedStrategy": return target.getConsumerProcessedStrategy();
+        case "consumerretrypause":
+        case "consumerRetryPause": return target.getConsumerRetryPause();
         case "delay": return target.getDelay();
         case "descending": return target.isDescending();
         case "designdocumentname":

components/camel-couchbase/src/generated/java/org/apache/camel/component/couchbase/CouchbaseEndpointConfigurer.java

@@ -23,7 +23,7 @@ public class CouchbaseEndpointUriFactory extends org.apache.camel.support.compon
     private static final Set<String> SECRET_PROPERTY_NAMES;
     private static final Set<String> MULTI_VALUE_PREFIXES;
     static {
-        Set<String> props = new HashSet<>(48);
+        Set<String> props = new HashSet<>(49);
         props.add("additionalHosts");
         props.add("autoStartIdForInserts");
         props.add("backoffErrorThreshold");
@@ -34,6 +34,7 @@ public class CouchbaseEndpointUriFactory extends org.apache.camel.support.compon
         props.add("collection");
         props.add("connectTimeout");
         props.add("consumerProcessedStrategy");
+        props.add("consumerRetryPause");
         props.add("delay");
         props.add("descending");
         props.add("designDocumentName");

components/camel-couchbase/src/generated/java/org/apache/camel/component/couchbase/CouchbaseEndpointUriFactory.java

@@ -76,9 +76,11 @@ protected static Boolean setDocument(
      * @param  queryTimeout
      * @return
      */
-    protected static GetResult getDocument(Collection collection, String id, long queryTimeout) {
+    protected static GetResult getDocument(Collection collection, String id, long queryTimeout, long retryPause) {
         GetOptions options = GetOptions.getOptions()
-                .timeout(Duration.ofMillis(queryTimeout));
+                .timeout(Duration.ofMillis(queryTimeout))
+                .retryStrategy(BestEffortRetryStrategy.withExponentialBackoff(Duration.ofMillis(retryPause),
+                        Duration.ofMillis(retryPause), 1));
         return collection.get(id, options);
     }
 

components/camel-couchbase/src/generated/resources/META-INF/org/apache/camel/component/couchbase/couchbase.json

@@ -137,7 +137,8 @@ protected int poll() throws Exception {
                 Object doc;
                 String id = row.id().get();
                 if (endpoint.isFullDocument()) {
-                    doc = CouchbaseCollectionOperation.getDocument(collection, id, endpoint.getQueryTimeout());
+                    doc = CouchbaseCollectionOperation.getDocument(collection, id, endpoint.getQueryTimeout(),
+                            endpoint.getConsumerRetryPause());
                 } else {
                     doc = row.valueAs(Object.class);
                 }
@@ -158,7 +159,7 @@ protected int poll() throws Exception {
                         LOG.trace("Deleting doc with ID {}", id);
                     }
                     CouchbaseCollectionOperation.removeDocument(collection, id, endpoint.getWriteQueryTimeout(),
-                            endpoint.getProducerRetryPause());
+                            endpoint.getConsumerRetryPause());
                 } else if ("filter".equalsIgnoreCase(consumerProcessedStrategy)) {
                     if (LOG.isTraceEnabled()) {
                         LOG.trace("Filtering out ID {}", id);

components/camel-couchbase/src/main/java/org/apache/camel/component/couchbase/CouchbaseCollectionOperation.java

@@ -131,6 +131,8 @@ public class CouchbaseEndpoint extends ScheduledPollEndpoint implements Endpoint
     private String rangeEndKey = "";
     @UriParam(label = "consumer", defaultValue = "false")
     private boolean fullDocument = true;
+    @UriParam(label = "consumer", defaultValue = "5000")
+    private int consumerRetryPause = DEFAULT_PAUSE_BETWEEN_RETRIES;
 
     // Consumer strategy
     @UriParam(label = "consumer", defaultValue = DEFAULT_CONSUME_PROCESSED_STRATEGY)
@@ -380,12 +382,23 @@ public int getProducerRetryPause() {
     }
 
     /**
-     * Define the retry pause between different attempts
+     * Define the producer retry pause between different attempts
      */
     public void setProducerRetryPause(int producerRetryPause) {
         this.producerRetryPause = producerRetryPause;
     }
 
+    public int getConsumerRetryPause() {
+        return consumerRetryPause;
+    }
+
+    /**
+     * Define the consumer retry pause between different attempts
+     */
+    public void setConsumerRetryPause(int consumerRetryPause) {
+        this.consumerRetryPause = consumerRetryPause;
+    }
+
     public String getDesignDocumentName() {
         return designDocumentName;
     }

components/camel-couchbase/src/main/java/org/apache/camel/component/couchbase/CouchbaseConsumer.java

@@ -143,7 +143,7 @@ public void process(Exchange exchange) throws Exception {
             exchange.getMessage().setBody(result);
         } else if (endpoint.getOperation().equals(COUCHBASE_GET)) {
             LOG.trace("Type of operation: GET");
-            GetResult result = CouchbaseCollectionOperation.getDocument(collection, id, queryTimeout);
+            GetResult result = CouchbaseCollectionOperation.getDocument(collection, id, queryTimeout, producerRetryPause);
             exchange.getMessage().setBody(result);
         } else if (endpoint.getOperation().equals(COUCHBASE_DELETE)) {
             LOG.trace("Type of operation: DELETE");

components/camel-couchbase/src/main/java/org/apache/camel/component/couchbase/CouchbaseEndpoint.java

@@ -116,6 +116,36 @@ default CouchbaseEndpointConsumerBuilder consumerProcessedStrategy(String consum
             doSetProperty("consumerProcessedStrategy", consumerProcessedStrategy);
             return this;
         }
+        /**
+         * Define the consumer retry pause between different attempts.
+         * 
+         * The option is a: <code>int</code> type.
+         * 
+         * Default: 5000
+         * Group: consumer
+         * 
+         * @param consumerRetryPause the value to set
+         * @return the dsl builder
+         */
+        default CouchbaseEndpointConsumerBuilder consumerRetryPause(int consumerRetryPause) {
+            doSetProperty("consumerRetryPause", consumerRetryPause);
+            return this;
+        }
+        /**
+         * Define the consumer retry pause between different attempts.
+         * 
+         * The option will be converted to a <code>int</code> type.
+         * 
+         * Default: 5000
+         * Group: consumer
+         * 
+         * @param consumerRetryPause the value to set
+         * @return the dsl builder
+         */
+        default CouchbaseEndpointConsumerBuilder consumerRetryPause(String consumerRetryPause) {
+            doSetProperty("consumerRetryPause", consumerRetryPause);
+            return this;
+        }
         /**
          * Define if this operation is descending or not.
          * 
@@ -1228,7 +1258,7 @@ default CouchbaseEndpointProducerBuilder producerRetryAttempts(String producerRe
             return this;
         }
         /**
-         * Define the retry pause between different attempts.
+         * Define the producer retry pause between different attempts.
          * 
          * The option is a: <code>int</code> type.
          * 
@@ -1243,7 +1273,7 @@ default CouchbaseEndpointProducerBuilder producerRetryPause(int producerRetryPau
             return this;
         }
         /**
-         * Define the retry pause between different attempts.
+         * Define the producer retry pause between different attempts.
          * 
          * The option will be converted to a <code>int</code> type.
          *