Keep multi-topic consumers running when a child consumer stops

src/Pulsar.Client/Api/PulsarClientException.fs

@@ -9,7 +9,9 @@ type IncompatibleSchemaException(msg) = inherit Exception(msg)
 type LookupException(msg) = inherit Exception(msg)
 type TooManyRequestsException(msg) = inherit Exception(msg)
 type ConnectException(msg) = inherit Exception(msg)
-type AlreadyClosedException(msg) = inherit Exception(msg)
+type AlreadyClosedException(msg: string, innerException: exn) =
+    inherit Exception(msg, innerException)
+    new(msg: string) = AlreadyClosedException(msg, null)
 type TopicTerminatedException(msg) = inherit Exception(msg)
 type AuthenticationException(msg) = inherit Exception(msg)
 type AuthorizationException(msg) = inherit Exception(msg)
@@ -19,7 +21,9 @@ type BrokerPersistenceException(msg) = inherit Exception(msg)
 type BrokerMetadataException(msg) = inherit Exception(msg)
 type ProducerBusyException(msg) = inherit Exception(msg)
 type ConsumerBusyException(msg) = inherit Exception(msg)
-type NotConnectedException(msg) = inherit Exception(msg)
+type NotConnectedException(msg: string, innerException: exn) =
+    inherit Exception(msg, innerException)
+    new(msg: string) = NotConnectedException(msg, null)
 type InvalidMessageException(msg) = inherit Exception(msg)
 type InvalidTopicNameException(msg) = inherit Exception(msg)
 type NotSupportedException(msg) = inherit Exception(msg)

src/Pulsar.Client/Internal/ConnectionHandler.fs

@@ -34,6 +34,8 @@ type internal ConnectionHandler( parentPrefix: string,
                         backoff: Backoff) as this =
 
     let mutable connectionState = Uninitialized
+    // why the handler moved to Failed, so later requests can report it
+    let mutable failureCause: exn = null
     let mutable lastDisconnectedTimestamp = 0L
     let mutable maxMessageSize = Commands.DEFAULT_MAX_MESSAGE_SIZE
     let mutable epoch = %0UL
@@ -106,7 +108,7 @@ type internal ConnectionHandler( parentPrefix: string,
             if t.IsFaulted then
                 let (Flatten ex) = t.Exception
                 Log.Logger.LogCritical(ex, "{0} mailbox failure", prefix)
-                this.ConnectionState <- Failed
+                this.Failed(ex)
             else
                 Log.Logger.LogInformation("{0} mailbox has stopped normally", prefix))
     |> ignore
@@ -125,7 +127,8 @@ type internal ConnectionHandler( parentPrefix: string,
     member this.Closed() =
         this.ConnectionState <- Closed
 
-    member this.Failed() =
+    member this.Failed(?cause: exn) =
+        failureCause <- defaultArg cause null
         this.ConnectionState <- Failed
 
     member this.Closing() =
@@ -151,12 +154,15 @@ type internal ConnectionHandler( parentPrefix: string,
         with get() : TimeStamp = %(Volatile.Read(&lastDisconnectedTimestamp))
         and private set(value: TimeStamp) = Volatile.Write(&lastDisconnectedTimestamp, %value)
 
+    member this.FailureCause = failureCause
+
     member this.CheckIfActive() =
         match this.ConnectionState with
         | Ready _ | Connecting -> null
         | Closing | Closed -> AlreadyClosedException(prefix + "already closed") :> exn
         | Terminated -> AlreadyClosedException(prefix + " topic was terminated") :> exn
-        | Failed | Uninitialized -> NotConnectedException(prefix + " not connected") :> exn
+        | Failed -> NotConnectedException(prefix + " not connected", failureCause) :> exn
+        | Uninitialized -> NotConnectedException(prefix + " not connected") :> exn
 
     member this.Close() =
         post mb Close

src/Pulsar.Client/Internal/ConsumerImpl.fs

@@ -584,7 +584,8 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien
 
     let stopConsumer () =
         closeConsumerTasks()
-        failWaiters <| AlreadyClosedException "Consumer is already closed"
+        // a failed consumer passes on why it stopped, e.g. to a multi-topic parent
+        failWaiters <| AlreadyClosedException("Consumer is already closed", connectionHandler.FailureCause)
         Log.Logger.LogInformation("{0} stopped", prefix)
 
     let isInvalidConsumerEpoch = isInvalidConsumerEpoch consumerConfig.SubscriptionType
@@ -904,7 +905,7 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien
     let stoppedMailboxException () =
         match connectionHandler.ConnectionState with
         | Closing | Closed | Terminated -> AlreadyClosedException(prefix + " already closed") :> exn
-        | _ -> NotConnectedException(prefix + " not connected") :> exn
+        | _ -> NotConnectedException(prefix + " not connected", connectionHandler.FailureCause) :> exn
 
     let failMailbox (ex: exn) =
         Log.Logger.LogCritical(ex, "{0} mailbox failure", prefix)
@@ -928,7 +929,7 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien
                 ()
         with closeEx ->
             Log.Logger.LogWarning(closeEx, "{0} failed to close consumer at the broker after mailbox failure", prefix)
-        connectionHandler.Failed()
+        connectionHandler.Failed(ex)
         stopConsumer()
         try
             currentMessage |> Option.iter (replyFromStoppedMailbox ex)
@@ -1017,7 +1018,7 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien
                             connectionHandler.ReconnectLater ex
                         | _ when not subscribeTsc.Task.IsCompleted ->
                             // unable to create new consumer, fail operation
-                            connectionHandler.Failed()
+                            connectionHandler.Failed(ex)
                             subscribeTsc.SetException(ex)
                             stopConsumer()
                             continueLoop <- false
@@ -1028,8 +1029,11 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien
                             //  * Regular consumer after topic is manually delete and with
                             //    auto-topic-creation set to false
                             // No more retries are needed in this case.
-                            connectionHandler.Failed()
+                            connectionHandler.Failed(ex)
                             closeConsumerTasks()
+                            // pending receives would otherwise wait forever, and a multi-topic parent
+                            // only learns that this child has stopped when its receive completes
+                            failWaiters <| NotConnectedException(prefix + " topic does not exist anymore", ex)
                             Log.Logger.LogWarning("{0} Closed consumer because topic does not exist anymore. {1}", prefix, ex.Message)
                             continueLoop <- false
                         | _ ->
@@ -1408,7 +1412,7 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien
                 if ((nonRetriableError || timeout) && subscribeTsc.TrySetException(ex)) then
                     Log.Logger.LogInformation("{0} creation failed {1}", prefix,
                                                 if nonRetriableError then "with unretriableError" else "after timeout")
-                    connectionHandler.Failed()
+                    connectionHandler.Failed(ex)
                     stopConsumer()
                     continueLoop <- false
 
@@ -1590,11 +1594,15 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien
         connectionHandler.GrabCnx()
         subscribeTsc.Task
 
+    // only before InitInternal, so that the first subscribe already uses it
+    member internal this.SetInitialConsumerEpoch(consumerEpoch: ConsumerEpoch) =
+        currentConsumerEpoch <- consumerEpoch
+
     static member Init(consumerConfig: ConsumerConfiguration<'T>, clientConfig: PulsarClientConfiguration,
                        topicName: TopicName, connectionPool: ConnectionPool, partitionIndex: int,
                        hasParent: bool, startMessageId: MessageId option, startMessageRollbackDuration: TimeSpan, lookup: ILookupService,
                        createTopicIfDoesNotExist: bool, schema: ISchema<'T>, schemaProvider: MultiVersionSchemaInfoProvider option,
-                       interceptors: ConsumerInterceptors<'T>, cleanup: ConsumerImpl<'T> -> unit) =
+                       interceptors: ConsumerInterceptors<'T>, cleanup: ConsumerImpl<'T> -> unit, ?consumerEpoch: ConsumerEpoch) =
         backgroundTask {
             let consumer =
                 if consumerConfig.ReceiverQueueSize > 0 then
@@ -1606,6 +1614,7 @@ type internal ConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clien
                                         startMessageId, lookup, createTopicIfDoesNotExist,
                                         schema, schemaProvider, interceptors, cleanup) :> ConsumerImpl<'T>
 
+            consumerEpoch |> Option.iter (fun epoch -> consumer.SetInitialConsumerEpoch epoch)
             do! consumer.InitInternal()
             return consumer
         }

src/Pulsar.Client/Internal/ConsumerInterceptors.fs

@@ -5,9 +5,14 @@ open Microsoft.Extensions.Logging
 open Pulsar.Client.Api
 open Pulsar.Client.Common
 
-type internal ConsumerInterceptors<'T>(interceptors: IConsumerInterceptor<'T> array) =
+type internal ConsumerInterceptors<'T>(interceptors: IConsumerInterceptor<'T> array, ownsInterceptors: bool) =
+     new(interceptors: IConsumerInterceptor<'T> array) = ConsumerInterceptors<'T>(interceptors, true)
      member this.Interceptors = interceptors
      static member Empty with get() = ConsumerInterceptors<'T>([||])
+
+     // the same interceptors for a consumer that doesn't own them, e.g. a child of a multi-topic consumer,
+     // which can stop and be replaced while the owner keeps using them
+     member this.Borrow() = ConsumerInterceptors<'T>(interceptors, false)
 
      member this.BeforeConsume (consumer: IConsumer<'T>, msg: Message<'T>) =
           let mutable interceptorMessage = msg         
@@ -50,8 +55,9 @@ type internal ConsumerInterceptors<'T>(interceptors: IConsumerInterceptor<'T> ar
                          Log.Logger.LogWarning(e, "Error executing interceptor OnAckTimeoutSend callback");
 
      member this.Close() =
-          for interceptor in interceptors do
-               try
-                    interceptor.Dispose()
-               with e ->
-                    Log.Logger.LogWarning(e, "Fail to close consumer interceptor");
+          if ownsInterceptors then
+               for interceptor in interceptors do
+                    try
+                         interceptor.Dispose()
+                    with e ->
+                         Log.Logger.LogWarning(e, "Fail to close consumer interceptor");