[Messaging]: MessageFlux: adding a retry const that limit retry to terminal completion event (Azure#45400)

* MessageFlux: adding a retry const that when used limit retry to terminal completion event

* Using completion based retry for event hubs

* update Service Bus pom to use core-amqp dev version

* minor rearrangement in version_client

* updating changelog for core-amqp and event hubs

Author: anuchandy

eng/versioning/version_client.txt

@@ -514,6 +514,8 @@ io.clientcore:annotation-processor-test;1.0.0-beta.1;1.0.0-beta.1
 # In the pom, the version update tag after the version should name the unreleased package and the dependency version:
 # <!-- {x-version-update;unreleased_com.azure:azure-core;dependency} -->
 
+unreleased_com.azure:azure-core-amqp;2.10.0-beta.1
+
 unreleased_io.clientcore:core;1.0.0-beta.10
 unreleased_io.clientcore:annotation-processor;1.0.0-beta.3
 unreleased_com.azure.v2:azure-core;2.0.0-beta.1

sdk/core/azure-core-amqp/CHANGELOG.md

@@ -4,6 +4,8 @@
 
 ### Features Added
 
+- Added an `AmqpRetryOptions` constant in `MessageFlux` to configure retries on terminal completion events but propagates terminal errors. ([45400](https://github.com/Azure/azure-sdk-for-java/pull/45400))
+
 ### Breaking Changes
 
 ### Bugs Fixed

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/MessageFlux.java

@@ -53,6 +53,11 @@ public final class MessageFlux extends FluxOperator<AmqpReceiveLink, Message> {
      * (i.e., disables the retry action to obtain next receiver from the upstream).
      **/
     public static final AmqpRetryPolicy NULL_RETRY_POLICY = new FixedAmqpRetryPolicy(new AmqpRetryOptions());
+    /** An AmqpRetryPolicy const indicates that MessageFlux should retry to obtain next receiver from the upstream
+     * only when the current receiver terminates with completion, any error from the current receiver will terminate
+     * the MessageFlux.
+     **/
+    public static final AmqpRetryPolicy RETRY_ONLY_COMPLETION = new FixedAmqpRetryPolicy(new AmqpRetryOptions());
     private static final String MESSAGE_FLUX_KEY = "messageFlux";
     private final ClientLogger logger;
     /**
@@ -650,6 +655,13 @@ private void setTerminationSignalOrScheduleNextMediatorRequest(Throwable error,
                 return;
             }
 
+            if (retryPolicy == RETRY_ONLY_COMPLETION) {
+                if (error != null) {
+                    onError(error);
+                    return;
+                }
+            }
+
             final Duration delay;
             if (error == null) {
                 // Even if the broker sets no link error condition, it's good to back off before creating a new link.

sdk/eventhubs/azure-messaging-eventhubs/CHANGELOG.md

@@ -4,6 +4,8 @@
 
 ### Features Added
 
+- Updated the partition receiver to retry on terminal completion signal instead of sending it downstream. ([45400](https://github.com/Azure/azure-sdk-for-java/pull/45400))
+
 ### Breaking Changes
 
 ### Bugs Fixed

sdk/eventhubs/azure-messaging-eventhubs/pom.xml

@@ -53,7 +53,7 @@
     <dependency>
       <groupId>com.azure</groupId>
       <artifactId>azure-core-amqp</artifactId>
-      <version>2.9.16</version> <!-- {x-version-update;com.azure:azure-core-amqp;dependency} -->
+      <version>2.10.0-beta.1</version> <!-- {x-version-update;unreleased_com.azure:azure-core-amqp;dependency} -->
     </dependency>
 
     <!-- Test dependencies -->

sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/EventHubConsumerAsyncClient.java

@@ -584,7 +584,7 @@ private EventHubPartitionAsyncConsumer createPartitionConsumer(String linkName,
         final MessageFluxWrapper linkMessageProcessor;
         if (connectionProcessor.isV2()) {
             MessageFlux messageFlux = new MessageFlux(receiveLinkFlux, prefetchCount, CreditFlowMode.EmissionDriven,
-                MessageFlux.NULL_RETRY_POLICY);
+                MessageFlux.RETRY_ONLY_COMPLETION);
             linkMessageProcessor
                 = new MessageFluxWrapper(InstrumentedMessageFlux.instrument(messageFlux, partitionId, instrumentation));
         } else {

sdk/eventhubs/azure-messaging-eventhubs/src/test/java/com/azure/messaging/eventhubs/EventHubPartitionAsyncConsumerTest.java

@@ -328,7 +328,7 @@ private static SystemProperties getSystemProperties(AmqpAnnotatedMessage amqpAnn
     private MessageFluxWrapper createLinkProcessor(boolean isV2) {
         if (isV2) {
             final MessageFlux messageFlux = new MessageFlux(createSink(link1, link2), PREFETCH,
-                CreditFlowMode.EmissionDriven, MessageFlux.NULL_RETRY_POLICY);
+                CreditFlowMode.EmissionDriven, MessageFlux.RETRY_ONLY_COMPLETION);
             return new MessageFluxWrapper(messageFlux);
         } else {
             final AmqpReceiveLinkProcessor receiveLinkProcessor

sdk/servicebus/azure-messaging-servicebus/pom.xml

@@ -72,7 +72,7 @@
     <dependency>
       <groupId>com.azure</groupId>
       <artifactId>azure-core-amqp</artifactId>
-      <version>2.9.16</version> <!-- {x-version-update;com.azure:azure-core-amqp;dependency} -->
+      <version>2.10.0-beta.1</version> <!-- {x-version-update;unreleased_com.azure:azure-core-amqp;dependency} -->
     </dependency>
     <dependency>
       <groupId>com.azure</groupId>