diff --git a/src/Pulsar.Client/Api/PulsarClientException.fs b/src/Pulsar.Client/Api/PulsarClientException.fs index 93ad3f97..4dfe532b 100644 --- a/src/Pulsar.Client/Api/PulsarClientException.fs +++ b/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) diff --git a/src/Pulsar.Client/Internal/ConnectionHandler.fs b/src/Pulsar.Client/Internal/ConnectionHandler.fs index 7a92ff9c..672721a8 100644 --- a/src/Pulsar.Client/Internal/ConnectionHandler.fs +++ b/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 \ No newline at end of file diff --git a/src/Pulsar.Client/Internal/ConsumerImpl.fs b/src/Pulsar.Client/Internal/ConsumerImpl.fs index 54908c57..7cc3d1b3 100644 --- a/src/Pulsar.Client/Internal/ConsumerImpl.fs +++ b/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 } diff --git a/src/Pulsar.Client/Internal/ConsumerInterceptors.fs b/src/Pulsar.Client/Internal/ConsumerInterceptors.fs index 33c08bbe..a7085f21 100644 --- a/src/Pulsar.Client/Internal/ConsumerInterceptors.fs +++ b/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"); \ No newline at end of file + if ownsInterceptors then + for interceptor in interceptors do + try + interceptor.Dispose() + with e -> + Log.Logger.LogWarning(e, "Fail to close consumer interceptor"); \ No newline at end of file diff --git a/src/Pulsar.Client/Internal/MultiTopicsConsumerImpl.fs b/src/Pulsar.Client/Internal/MultiTopicsConsumerImpl.fs index 97fa8cbb..4c8c5e34 100644 --- a/src/Pulsar.Client/Internal/MultiTopicsConsumerImpl.fs +++ b/src/Pulsar.Client/Internal/MultiTopicsConsumerImpl.fs @@ -40,9 +40,30 @@ type internal BatchAddResponse<'T> = type internal PolledMessage<'T> = (struct (CancellationToken * ResultOrException>)) +type internal ChildConsumerInfo<'T> = + { + // the subscribed topic, as tracked in allTopics + TopicName: TopicName + // the partition for partitioned topics, otherwise the topic itself + ChildTopicName: TopicName + IsPartitioned: bool + // takes the parent's consumer epoch, otherwise the parent discards the new child's messages as stale + Create: ConsumerEpoch -> Task> + } + +type internal ChildConsumer<'T> = + { + Info: ChildConsumerInfo<'T> + Consumer: IConsumer<'T> + Stream: TaskGenerator> + } + type internal MultiTopicConsumerMessage<'T> = | Init | PollerFailed of exn + | ChildStopped of CompleteTopicName * IConsumer<'T> * exn * detached: TaskCompletionSource + | RecreateChild of ChildConsumerInfo<'T> * Backoff + | ChildRecreated of ChildConsumerInfo<'T> * Task> * ConsumerEpoch * Backoff | Receive of ReceiveCallback<'T> | BatchReceive of ReceiveCallbacks<'T> | MessageReceived of PolledMessage<'T> * TaskCompletionSource @@ -68,13 +89,6 @@ type internal MultiTopicConsumerMessage<'T> = | IsConnected of TaskCompletionSource | HasMessageAvailable of TaskCompletionSource -type internal TopicAndConsumer<'T> = - { - IsPartitioned: bool - TopicName: TopicName - Consumer: ConsumerImpl<'T> - } - type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration<'T>, clientConfig: PulsarClientConfiguration, connectionPool: ConnectionPool, multiConsumerType: MultiConsumerType<'T>, startMessageId: MessageId option, startMessageRollbackDuration: TimeSpan, lookup: ILookupService, @@ -84,13 +98,23 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration let consumerId = Generators.getNextConsumerId() let consumerName = getConsumerName consumerConfig.ConsumerName let prefix = $"mt/consumer({consumerId}, {consumerName})" - let consumers = Dictionary * TaskGenerator>>() + let consumers = Dictionary>() + // children that stopped on their own and are waiting to be recreated + let pendingChildren = Dictionary>() + // creations of those children that are still running; ChildRecreated takes them over, unless Close does first + let recreatingChildren = HashSet>>() + // children being disposed without the mailbox waiting for them + let childDisposals = ResizeArray() + // children stop and are replaced while the parent keeps running, so only the parent disposes the interceptors + let childInterceptors = interceptors.Borrow() let consumerCreatedTsc = TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously) let pollerCts = new CancellationTokenSource() let mutable receiveCts = new CancellationTokenSource() // Late poller completions can restart a generator after shutdown disposes the source. let mutable receiveToken = receiveCts.Token let mutable connectionState = MultiTopicConnectionState.Uninitialized + // why the consumer failed, so requests made after it stopped can report it + let mutable terminalError: exn = null let mutable currentStream = Unchecked.defaultof>> let partitionedTopics = Dictionary>() let allTopics = HashSet() @@ -192,11 +216,21 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration if hasReachedEndOfTopic then Log.Logger.LogWarning("{0} topic was terminated", topic) do! Task.Delay(Timeout.Infinite, cancellationToken) - let! message = consumer.ReceiveWrappedAsync(cancellationToken) + let! message = + try + consumer.ReceiveWrappedAsync(cancellationToken) + with ex -> + // a stopped child throws before it posts the receive + Task.FromResult(Error ex) match message with | Error ex when (ex :? AlreadyClosedException) || (ex :? NotConnectedException) -> - // Stop the poller before a terminal child failure can be buffered as a message. - return reraize ex + // A stopped child can't deliver anything else. The parent detaches it instead of buffering + // the error as a message or failing the poller, and releases this read once its generator + // is gone, so the result is never polled. + let detached = TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously) + post this.Mb (ChildStopped (topic, consumerImp, ex, detached)) + do! detached.Task.WaitAsync(cancellationToken) + return struct (cancellationToken, Error ex) | _ -> let message = message |> Result.map (fun msg -> let newMessageId = { msg.MessageId with TopicName = topic } @@ -210,21 +244,21 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration let resetStreams (action: IConsumer<'T> -> Task) = backgroundTask { let children = consumers.Values |> Seq.toArray - for _, stream in children do - do! currentStream.RemoveGenerator stream + for child in children do + do! currentStream.RemoveGenerator child.Stream // Stop prefetch before moving any cursor, including reads whose results are already queued. receiveCts.Cancel() try let! _ = children - |> Array.map (fun (consumer, _) -> backgroundTask { do! action consumer }) + |> Array.map (fun child -> backgroundTask { do! action child.Consumer }) |> Task.WhenAll () finally receiveCts.Dispose() receiveCts <- new CancellationTokenSource() receiveToken <- receiveCts.Token - currentStream.AddGenerators (children |> Array.map snd) + currentStream.AddGenerators (children |> Array.map _.Stream) } let stopConsumer() = @@ -236,45 +270,116 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration pollerCts.Dispose() receiveCts.Cancel() receiveCts.Dispose() + interceptors.Close() + let closedException = AlreadyClosedException("Consumer is already closed", terminalError) while waiters.Count > 0 do let waitingChannel = waiters |> dequeueWaiter - AlreadyClosedException "Consumer is already closed" |> waitingChannel.TrySetException |> ignore + closedException |> waitingChannel.TrySetException |> ignore while batchWaiters.Count > 0 do let batchWaitingChannel = batchWaiters |> dequeueBatchWaiter - AlreadyClosedException "Consumer is already closed" |> batchWaitingChannel.TrySetException |> ignore + closedException |> batchWaitingChannel.TrySetException |> ignore Log.Logger.LogInformation("{0} stopped", prefix) + let failConsumer (ex: exn) = + terminalError <- ex + this.ConnectionState <- Failed + stopConsumer() + + let childConsumerInfo (consumerInitInfo: ConsumerInitInfo<'T>) (childTopic: TopicName) partitionIndex + receiverQueueSize createTopicIfDoesNotExist = + let childConfig = { consumerConfig with + ReceiverQueueSize = receiverQueueSize + Topics = [| childTopic |] } + { + TopicName = consumerInitInfo.TopicName + ChildTopicName = childTopic + IsPartitioned = partitionIndex >= 0 + Create = fun consumerEpoch -> + ConsumerImpl.Init(childConfig, clientConfig, childConfig.SingleTopic, + connectionPool, partitionIndex, true, startMessageId, startMessageRollbackDuration, + lookup, createTopicIfDoesNotExist, consumerInitInfo.Schema, consumerInitInfo.SchemaProvider, + childInterceptors, (fun _ -> ()), consumerEpoch = consumerEpoch) + } + + // only called from the mailbox, so the children are never changed concurrently + let registerChild (info: ChildConsumerInfo<'T>) (consumer: ConsumerImpl<'T>) = + let childTopic = info.ChildTopicName.CompleteTopicName + let stream = getStream childTopic consumer + consumers.Add(childTopic, { Info = info; Consumer = consumer :> IConsumer<'T>; Stream = stream }) + stream + + // the topic was removed, or its consumer stopped and hasn't been recreated yet + let noChildException (topic: CompleteTopicName) = + NotConnectedException($"{prefix} has no consumer for {topic}") :> exn + + // Seeking or unsubscribing now would silently skip the children being recreated: they would keep their old + // position, or their subscription. + let pendingChildrenException (operation: string) = + let topics = String.Join(", ", pendingChildren.Keys) + NotConnectedException($"{prefix} can't {operation} while the consumers for {topics} are being recreated") :> exn + + // doesn't block the mailbox, but Close still waits for it in closeChildren + let disposeChild (consumer: IConsumer<'T>) = + childDisposals.RemoveAll(fun disposal -> disposal.IsCompleted) |> ignore + let disposal = + backgroundTask { + try + do! consumer.DisposeAsync() + with Flatten ex -> + Log.Logger.LogWarning(ex, "{0} could not dispose a child consumer", prefix) + } + childDisposals.Add disposal + + let disposeCreatedChildren (consumersTasks: Task> seq) = + consumersTasks + |> Seq.filter (fun t -> t.Status = TaskStatus.RanToCompletion) + |> Seq.map (fun t -> (t.Result :> IConsumer<'T>).DisposeAsync().AsTask()) + |> Task.WhenAll + + let disposeRecreatingChild (createTask: Task>) = + backgroundTask { + // a failed creation has nothing to close + let! created = createTask.ContinueWith(fun (t: Task>) -> t.IsCompletedSuccessfully) + if created then + do! (createTask.Result :> IConsumer<'T>).DisposeAsync() + } :> Task + + // Also waits for the children still being recreated or disposed, otherwise they could stay subscribed after + // Close completes. + let closeChildren () = + let recreating = recreatingChildren |> Seq.toArray + recreatingChildren.Clear() + let disposals = childDisposals.ToArray() + childDisposals.Clear() + consumers.Values + |> Seq.map (fun child -> child.Consumer.DisposeAsync().AsTask()) + |> Seq.append (recreating |> Array.map disposeRecreatingChild) + |> Seq.append disposals + |> Task.WhenAll + + let createChildBackoff () = + Backoff { BackoffConfig.Default with + Initial = clientConfig.InitialBackoffInterval + Max = clientConfig.MaxBackoffInterval } + + let scheduleChildRecreation (info: ChildConsumerInfo<'T>) (backoff: Backoff) = + let delay = backoff.Next() + Log.Logger.LogInformation("{0} will recreate the consumer for {1} in {2} ms", prefix, info.ChildTopicName, delay) + asyncDelayMs delay (fun () -> post this.Mb (RecreateChild (info, backoff))) + let singleInit (consumerInitInfo: ConsumerInitInfo<'T>) = let topic = consumerInitInfo.TopicName let numPartitions = consumerInitInfo.Metadata.Partitions let receiverQueueSize = Math.Min(consumerConfig.ReceiverQueueSize, consumerConfig.MaxTotalReceiverQueueSizeAcrossPartitions / numPartitions) - let consumersTasks = - Seq.init numPartitions (fun partitionIndex -> - let partitionedTopic = topic.GetPartition(partitionIndex) - let partititonedConfig = { consumerConfig with - ReceiverQueueSize = receiverQueueSize - Topics = [| partitionedTopic |] } - backgroundTask { - let! result = - ConsumerImpl.Init(partititonedConfig, clientConfig, partititonedConfig.SingleTopic, - connectionPool, partitionIndex, true, startMessageId, startMessageRollbackDuration, - lookup, true, consumerInitInfo.Schema, consumerInitInfo.SchemaProvider, - interceptors, fun _ -> ()) - return { IsPartitioned = true; TopicName = partitionedTopic; Consumer = result } - }) - |> Seq.cache + let infos = + Array.init numPartitions (fun partitionIndex -> + childConsumerInfo consumerInitInfo (topic.GetPartition partitionIndex) partitionIndex receiverQueueSize true) + let consumersTasks = infos |> Array.map (fun info -> info.Create currentConsumerEpoch) backgroundTask { let! consumerResults = consumersTasks |> Task.WhenAll - allTopics.Add consumerInitInfo.TopicName |> ignore + allTopics.Add topic |> ignore partitionedTopics.Add(topic, consumerInitInfo) - return - consumerResults - |> Seq.map (fun topicAndConsumer -> - let stream = getStream topicAndConsumer.TopicName.CompleteTopicName topicAndConsumer.Consumer - consumers.Add(topicAndConsumer.TopicName.CompleteTopicName, (topicAndConsumer.Consumer :> IConsumer<'T>, stream)) - stream - ) - |> Seq.cache + return Array.map2 registerChild infos consumerResults }, consumersTasks let multiInit (consumerInitInfos: ConsumerInitInfo<'T>[]) createTopicIfDoesNotExist = @@ -291,56 +396,26 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration else totalConsumersCount <- totalConsumersCount + 1 let receiverQueueSize = Math.Min(consumerConfig.ReceiverQueueSize, consumerConfig.MaxTotalReceiverQueueSizeAcrossPartitions / totalConsumersCount) - let consumersTasks = + let infos = consumerInitInfos - |> Seq.collect - (fun consumerInitInfo -> - if consumerInitInfo.Metadata.IsMultiPartitioned then - let topic = consumerInitInfo.TopicName - let numPartitions = consumerInitInfo.Metadata.Partitions - Seq.init numPartitions (fun partitionIndex -> - let partitionedTopic = topic.GetPartition(partitionIndex) - let partititonedConfig = { consumerConfig with - ReceiverQueueSize = receiverQueueSize - Topics = [| partitionedTopic |] } - backgroundTask { - let! result = - ConsumerImpl.Init(partititonedConfig, clientConfig, partititonedConfig.SingleTopic, - connectionPool, partitionIndex, true, startMessageId, startMessageRollbackDuration, - lookup, createTopicIfDoesNotExist, consumerInitInfo.Schema, consumerInitInfo.SchemaProvider, - interceptors, fun _ -> ()) - return { IsPartitioned = true; TopicName = partitionedTopic; Consumer = result } - }) - else - seq { - backgroundTask { - let partititonedConfig = { consumerConfig with - ReceiverQueueSize = receiverQueueSize - Topics = [| consumerInitInfo.TopicName |] } - let! result = - ConsumerImpl.Init(partititonedConfig, clientConfig, partititonedConfig.SingleTopic, - connectionPool, -1, true, startMessageId, startMessageRollbackDuration, - lookup, createTopicIfDoesNotExist, consumerInitInfo.Schema, consumerInitInfo.SchemaProvider, - interceptors, fun _ -> ()) - return { IsPartitioned = false; TopicName = consumerInitInfo.TopicName; Consumer = result } - } - }) - |> Seq.cache + |> Array.collect (fun consumerInitInfo -> + let topic = consumerInitInfo.TopicName + if consumerInitInfo.Metadata.IsMultiPartitioned then + Array.init consumerInitInfo.Metadata.Partitions (fun partitionIndex -> + childConsumerInfo consumerInitInfo (topic.GetPartition partitionIndex) partitionIndex + receiverQueueSize createTopicIfDoesNotExist) + else + [| childConsumerInfo consumerInitInfo topic (-1) receiverQueueSize createTopicIfDoesNotExist |]) + let consumersTasks = infos |> Array.map (fun info -> info.Create currentConsumerEpoch) backgroundTask { let! consumerResults = consumersTasks |> Task.WhenAll allTopics.UnionWith newTopics for consumerInfo in newPartitionedConsumers do partitionedTopics.Add(consumerInfo.TopicName, consumerInfo) - return - consumerResults - |> Seq.map (fun topicAndConsumer -> - let stream = getStream topicAndConsumer.TopicName.CompleteTopicName topicAndConsumer.Consumer - consumers.Add(topicAndConsumer.TopicName.CompleteTopicName, (topicAndConsumer.Consumer :> IConsumer<'T>, stream)) - stream) - |> Seq.cache + return Array.map2 registerChild infos consumerResults }, consumersTasks else - Task.FromResult(Seq.empty), Seq.empty + Task.FromResult([||]), [||] let processAddedTopics (topicsToAdd: TopicName seq) (getConsumerInitInfo: TopicName -> Task>) = backgroundTask { @@ -353,50 +428,31 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration return! addedTopicsTask with Flatten ex -> Log.Logger.LogError(ex, "{0} could not processAddedTopics", prefix) - do! consumersTasks - |> Seq.filter (fun t -> t.Status = TaskStatus.RanToCompletion) - |> Seq.map (fun t -> - let consumer = t.Result.Consumer - (consumer :> IConsumer<'T>).DisposeAsync().AsTask()) - |> Task.WhenAll - return Seq.empty + do! disposeCreatedChildren consumersTasks + return [||] } - let processRemovedTopics (topicsToRemove: TopicName seq) = - let consumersTasks = - topicsToRemove - |> Seq.collect(fun topicToRemove -> - consumers - |> Seq.filter(fun (KeyValue(topic, _)) -> - let t: string = %topic - let lastIndexOfPartition = t.LastIndexOf("-partition-") - topicToRemove.CompleteTopicName = - if lastIndexOfPartition > 0 then - %t.Substring(0, lastIndexOfPartition) - else - %t) - |> Seq.map(fun (KeyValue(topic, (consumer, stream))) -> backgroundTask { - do! currentStream.RemoveGenerator stream - do! consumer.DisposeAsync() - return topicToRemove, topic - }) - ) - |> Seq.cache + let processRemovedTopics (topicsToRemove: HashSet) = backgroundTask { - let! removedTopics = backgroundTask { - try - return! consumersTasks |> Task.WhenAll - with Flatten ex -> - Log.Logger.LogError(ex, "{0} could not processRemovedTopics fully", prefix) - return consumersTasks - |> Seq.filter (fun t -> t.Status = TaskStatus.RanToCompletion) - |> Seq.map _.Result - |> Seq.toArray - } - for removedTopic, removedTopicPartition in removedTopics do - consumers.Remove(removedTopicPartition) |> ignore + let removedChildren = + consumers + |> Seq.filter (fun (KeyValue(_, child)) -> topicsToRemove.Contains child.Info.TopicName) + |> Seq.toArray + for KeyValue(childTopic, child) in removedChildren do + do! currentStream.RemoveGenerator child.Stream + consumers.Remove(childTopic) |> ignore + for KeyValue(childTopic, info) in pendingChildren |> Seq.toArray do + if topicsToRemove.Contains info.TopicName then + pendingChildren.Remove(childTopic) |> ignore + for removedTopic in topicsToRemove do allTopics.Remove(removedTopic) |> ignore partitionedTopics.Remove(removedTopic) |> ignore + try + do! removedChildren + |> Array.map (fun (KeyValue(_, child)) -> child.Consumer.DisposeAsync().AsTask()) + |> Task.WhenAll + with Flatten ex -> + Log.Logger.LogError(ex, "{0} could not processRemovedTopics fully", prefix) } let isPollingAllowed() = @@ -483,8 +539,8 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration let! newPartitionsOption = getAllPartitions() match newPartitionsOption with | Some newPartitions -> - let oldConsumersCount = consumers.Count - let mutable totalConsumersCount = oldConsumersCount + // children being recreated still count towards the receiver queue split + let oldConsumersCount = consumers.Count + pendingChildren.Count let topicsToUpdate = newPartitions |> Array.filter(fun (topic, partitionedTopicNames) -> @@ -503,61 +559,43 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration false) if topicsToUpdate.Length > 0 then Log.Logger.LogInformation("{0} adding subscription to {1} new partitions", prefix, topicsToUpdate.Length) + // counted with the new partitions, so it is never zero + let totalConsumersCount = + topicsToUpdate + |> Array.sumBy (fun (topic, partitionedTopicNames) -> + partitionedTopicNames.Length - partitionedTopics[topic].Metadata.Partitions) + |> (+) oldConsumersCount let receiverQueueSize = Math.Min(consumerConfig.ReceiverQueueSize, consumerConfig.MaxTotalReceiverQueueSizeAcrossPartitions / totalConsumersCount) - let newConsumerTasks = - seq { + let infos = + [| for topic, partitionedTopicNames in topicsToUpdate do let consumerInitInfo = partitionedTopics[topic] let oldPartitionsCount = consumerInitInfo.Metadata.Partitions let newPartitionsCount = partitionedTopicNames.Length - totalConsumersCount <- totalConsumersCount + (newPartitionsCount - oldPartitionsCount) - partitionedTopics[topic] <- - { - consumerInitInfo with - Metadata = { - Partitions = newPartitionsCount - } - } - let newConsumerTasks = - seq { oldPartitionsCount..newPartitionsCount - 1 } - |> Seq.map (fun partitionIndex -> - let partitionedTopic = partitionedTopicNames[partitionIndex] - let partititonedConfig = { consumerConfig with - ReceiverQueueSize = receiverQueueSize - Topics = [| partitionedTopic |] } - backgroundTask { - let! result = - ConsumerImpl.Init(partititonedConfig, clientConfig, partititonedConfig.SingleTopic, - connectionPool, partitionIndex, true, startMessageId, startMessageRollbackDuration, - lookup, true, consumerInitInfo.Schema, consumerInitInfo.SchemaProvider, - interceptors, fun _ -> ()) - return (partitionedTopic, result) - }) - yield! newConsumerTasks - } + for partitionIndex in oldPartitionsCount..newPartitionsCount - 1 do + let partitionedTopic = partitionedTopicNames[partitionIndex] + childConsumerInfo consumerInitInfo partitionedTopic partitionIndex receiverQueueSize true + |] + let newConsumerTasks = infos |> Array.map (fun info -> info.Create currentConsumerEpoch) try - let! newConsumerResults = - newConsumerTasks - |> Task.WhenAll - let newStreams = - newConsumerResults - |> Seq.map (fun (topic, consumer) -> - let stream = getStream topic.CompleteTopicName consumer - consumers.Add(topic.CompleteTopicName, (consumer :> IConsumer<'T>, stream)) - stream) - |> Seq.cache + let! newConsumers = newConsumerTasks |> Task.WhenAll + // the new partitions are only recorded once all of them have consumers, so a failed attempt is retried + for topic, partitionedTopicNames in topicsToUpdate do + partitionedTopics[topic] <- + { + partitionedTopics[topic] with + Metadata = { + Partitions = partitionedTopicNames.Length + } + } + let newStreams = Array.map2 registerChild infos newConsumers currentStream.AddGenerators(newStreams) Log.Logger.LogDebug("{0} success create consumers for extended partitions. old: {1}, new: {2}", prefix, oldConsumersCount, totalConsumersCount ) with Flatten ex -> Log.Logger.LogWarning(ex, "{0} fail create consumers for extended partitions. old: {1}, new: {2}", prefix, oldConsumersCount, totalConsumersCount ) - do! newConsumerTasks - |> Seq.filter (fun t -> t.Status = TaskStatus.RanToCompletion) - |> Seq.map (fun t -> - let _, consumer = t.Result - (consumer :> IConsumer<'T>).DisposeAsync().AsTask()) - |> Task.WhenAll + do! disposeCreatedChildren newConsumerTasks Log.Logger.LogInformation("{0} disposed partially created consumers", prefix) | None -> @@ -696,15 +734,15 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration channel.TrySetCanceled() |> ignore | CancelBatchWaiter (struct (_, _, channel)) -> channel.TrySetCanceled() |> ignore + | ChildRecreated (_, createTask, _, _) -> + // unless Close has already taken it over + if recreatingChildren.Remove createTask && createTask.IsCompletedSuccessfully then + disposeChild createTask.Result | Close channel -> // the child consumers outlive a failed mailbox, so they still have to be closed backgroundTask { try - let! _ = - consumers - |> Seq.map (fun (KeyValue(_, (consumer, _))) -> consumer.DisposeAsync().AsTask()) - |> Task.WhenAll - () + do! closeChildren() with Flatten closeEx -> Log.Logger.LogError(closeEx, "{0} could not close all child consumers properly", prefix) this.ConnectionState <- Closed @@ -716,7 +754,10 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration let stoppedMailboxException () = match this.ConnectionState with | Closing | Closed -> AlreadyClosedException(prefix + " already closed") :> exn - | _ -> NotConnectedException(prefix + " not connected") :> exn + | _ -> + match terminalError with + | null -> NotConnectedException(prefix + " not connected") :> exn + | cause -> NotConnectedException($"{prefix} has failed and can't recover, it has to be recreated: {cause.Message}", cause) :> exn // requests posted after the mailbox exits would otherwise never be completed let drainStoppedMailbox () = @@ -758,25 +799,96 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration consumerCreatedTsc.SetResult() with Flatten ex -> Log.Logger.LogError(ex, "{0} could not create", prefix) - do! consumersTasks - |> Seq.filter (fun t -> t.Status = TaskStatus.RanToCompletion) - |> Seq.map (fun t -> - let consumer = t.Result.Consumer - (consumer :> IConsumer<'T>).DisposeAsync().AsTask()) - |> Task.WhenAll - this.ConnectionState <- Failed + do! disposeCreatedChildren consumersTasks + failConsumer ex consumerCreatedTsc.SetException(ex) - stopConsumer() if this.ConnectionState = Failed then continueLoop <- false | PollerFailed ex -> Log.Logger.LogError(ex, "{0} stopping because its poller failed", prefix) - this.ConnectionState <- Failed - stopConsumer() + failConsumer ex continueLoop <- false + | ChildStopped (childTopic, consumer, ex, detached) -> + + match consumers.TryGetValue childTopic with + | true, child when Object.ReferenceEquals(child.Consumer, consumer) -> + match this.ConnectionState with + | Ready -> + Log.Logger.LogWarning(ex, "{0} consumer for {1} has stopped", prefix, childTopic) + do! currentStream.RemoveGenerator child.Stream + consumers.Remove(childTopic) |> ignore + disposeChild child.Consumer + match multiConsumerType with + | Pattern _ when consumerConfig.SubscriptionMode = SubscriptionMode.NonDurable + && not child.Info.IsPartitioned -> + // A recreated child would start again from the initial position, like any topic that + // discovery adds, so the next discovery subscribes again if the topic still matches. + allTopics.Remove(child.Info.TopicName) |> ignore + | _ when consumerConfig.SubscriptionMode = SubscriptionMode.NonDurable -> + // a recreated child would start again from the initial position, so it can't take over + Log.Logger.LogError(ex, "{0} stopping because the non-durable consumer for {1} has stopped", prefix, childTopic) + failConsumer ex + continueLoop <- false + | _ -> + // A deleted pattern topic can't be subscribed again, so it stays pending until discovery + // removes it. Seek and Unsubscribe refuse until then instead of skipping the topic. + pendingChildren[childTopic] <- child.Info + scheduleChildRecreation child.Info (createChildBackoff()) + // the child's generator has been removed, so its parked read can finish + detached.TrySetResult() |> ignore + | Failed -> + // Only a failed Unsubscribe leaves the mailbox running in this state. It may have unsubscribed + // this child already, so the child can't be recreated; stopping cancels its parked read. + Log.Logger.LogError(ex, "{0} stopping because the consumer for {1} has stopped after a failed unsubscribe", prefix, childTopic) + failConsumer ex + continueLoop <- false + | _ -> + // stopping the consumer cancels the parked read + Log.Logger.LogDebug("{0} ignored the stopped consumer for {1} in state {2}", prefix, childTopic, this.ConnectionState) + | _ -> + Log.Logger.LogDebug("{0} ignored the stopped consumer for {1}", prefix, childTopic) + // its generator has already been removed, so its parked read can finish + detached.TrySetResult() |> ignore + + | RecreateChild (info, backoff) -> + + let childTopic = info.ChildTopicName.CompleteTopicName + if this.ConnectionState = Ready && pendingChildren.ContainsKey childTopic then + Log.Logger.LogInformation("{0} recreating the consumer for {1}", prefix, childTopic) + let consumerEpoch = currentConsumerEpoch + // completes off the mailbox, so other topics keep being served in the meantime + let createTask = info.Create consumerEpoch + recreatingChildren.Add createTask |> ignore + createTask.ContinueWith(fun (_: Task>) -> + post this.Mb (ChildRecreated (info, createTask, consumerEpoch, backoff))) |> ignore + + | ChildRecreated (info, createTask, consumerEpoch, backoff) -> + + let childTopic = info.ChildTopicName.CompleteTopicName + if recreatingChildren.Remove createTask then + if createTask.IsCompletedSuccessfully then + let consumer = createTask.Result + // the topic may have been removed while the child was being created + if this.ConnectionState = Ready && pendingChildren.ContainsKey childTopic && not (consumers.ContainsKey childTopic) then + if consumerEpoch = currentConsumerEpoch then + pendingChildren.Remove(childTopic) |> ignore + let stream = registerChild info consumer + currentStream.AddGenerators [| stream |] + Log.Logger.LogInformation("{0} recreated the consumer for {1}", prefix, childTopic) + else + // redelivery moved the epoch on while the child was created, so its messages would be discarded + disposeChild consumer + scheduleChildRecreation info (createChildBackoff()) + else + disposeChild consumer + else + let (Flatten ex) = createTask.Exception + Log.Logger.LogWarning(ex, "{0} could not recreate the consumer for {1}", prefix, childTopic) + scheduleChildRecreation info backoff + | MessageReceived (struct (cancellationToken, message), pollerChannel) -> let hasWaitingChannel = waiters.Count > 0 @@ -828,49 +940,63 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration | Acknowledge (channel, msgId, txnOption) -> Log.Logger.LogDebug("{0} Acknowledge {1}", prefix, msgId) - let consumer, _ = consumers[msgId.TopicName] - backgroundTask { - try - match txnOption with - | Some txn -> - do! consumer.AcknowledgeAsync(msgId, txn) - | None -> - do! consumer.AcknowledgeAsync(msgId) - unAckedMessageTracker.Remove msgId - channel.SetResult() - with Flatten ex -> - channel.SetException(ex) - } |> ignore + match consumers.TryGetValue msgId.TopicName with + | true, child -> + backgroundTask { + try + match txnOption with + | Some txn -> + do! child.Consumer.AcknowledgeAsync(msgId, txn) + | None -> + do! child.Consumer.AcknowledgeAsync(msgId) + unAckedMessageTracker.Remove msgId + channel.SetResult() + with Flatten ex -> + channel.SetException(ex) + } |> ignore + | _ -> + // interceptors are told, as by a single-topic consumer, e.g. so tracing ends for the message + let ex = noChildException msgId.TopicName + interceptors.OnAcknowledge(this, msgId, ex) + channel.SetException ex | NegativeAcknowledge (channel, msgId) -> Log.Logger.LogDebug("{0} NegativeAcknowledge {1}", prefix, msgId) - let consumer, _ = consumers[msgId.TopicName] - backgroundTask { - try - do! consumer.NegativeAcknowledge msgId - unAckedMessageTracker.Remove msgId - channel.SetResult() - with Flatten ex -> - channel.SetException ex - } |> ignore + match consumers.TryGetValue msgId.TopicName with + | true, child -> + backgroundTask { + try + do! child.Consumer.NegativeAcknowledge msgId + unAckedMessageTracker.Remove msgId + channel.SetResult() + with Flatten ex -> + channel.SetException ex + } |> ignore + | _ -> + channel.SetException(noChildException msgId.TopicName) | AcknowledgeCumulative (channel, msgId, txnOption) -> Log.Logger.LogDebug("{0} AcknowledgeCumulative {1}", prefix, msgId) - let consumer, _ = consumers[msgId.TopicName] - backgroundTask { - try - match txnOption with - | Some txn -> - do! consumer.AcknowledgeCumulativeAsync(msgId, txn) - | None -> - do! consumer.AcknowledgeCumulativeAsync msgId - unAckedMessageTracker.RemoveMessagesTill msgId |> ignore - channel.SetResult() - with Flatten ex -> - channel.SetException ex - } |> ignore + match consumers.TryGetValue msgId.TopicName with + | true, child -> + backgroundTask { + try + match txnOption with + | Some txn -> + do! child.Consumer.AcknowledgeCumulativeAsync(msgId, txn) + | None -> + do! child.Consumer.AcknowledgeCumulativeAsync msgId + unAckedMessageTracker.RemoveMessagesTill msgId |> ignore + channel.SetResult() + with Flatten ex -> + channel.SetException ex + } |> ignore + | _ -> + let ex = noChildException msgId.TopicName + interceptors.OnAcknowledgeCumulative(this, msgId, ex) + channel.SetException ex | RedeliverAllUnacknowledged channel -> @@ -903,8 +1029,12 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration messageIds |> Seq.groupBy _.TopicName |> Seq.iter(fun (topicName, msgIds) -> - let consumer, _ = consumers[topicName] - msgIds |> RedeliverSet |> (consumer :?> ConsumerImpl<'T>).RedeliverUnacknowledged + match consumers.TryGetValue topicName with + | true, child -> + msgIds |> RedeliverSet |> (child.Consumer :?> ConsumerImpl<'T>).RedeliverUnacknowledged + | _ -> + // the broker redelivers them to a recreated consumer anyway + Log.Logger.LogDebug("{0} no consumer to redeliver messages for {1}", prefix, topicName) ) with ex -> Log.Logger.LogError(ex, "{0} RedeliverUnacknowledgedMessages failed", prefix) @@ -939,53 +1069,61 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration | HasReachedEndOfTheTopic channel -> Log.Logger.LogDebug("{0} HasReachedEndOfTheTopic", prefix) - consumers - |> Seq.forall (fun (KeyValue(_, (consumer, _))) -> consumer.HasReachedEndOfTopic().Result) + // a child being recreated hasn't stopped because its topic was terminated + (pendingChildren.Count = 0 + && consumers |> Seq.forall (fun (KeyValue(_, child)) -> child.Consumer.HasReachedEndOfTopic().Result)) |> channel.SetResult | LastDisconnectedTimestamp channel -> Log.Logger.LogDebug("{0} LastDisconnectedTimestamp", prefix) + // there may be no children, e.g. when a pattern matches no topics consumers - |> Seq.map (fun (KeyValue(_, (consumer, _))) -> consumer.LastDisconnectedTimestamp().Result) - |> Seq.max + |> Seq.map (fun (KeyValue(_, child)) -> child.Consumer.LastDisconnectedTimestamp().Result) + |> Seq.fold max %0L |> channel.SetResult | IsConnected channel -> Log.Logger.LogDebug("{0} IsConnected", prefix) - consumers - |> Seq.forall (fun (KeyValue(_, (consumer, _))) -> consumer.IsConnected().Result) + (pendingChildren.Count = 0 + && consumers |> Seq.forall (fun (KeyValue(_, child)) -> child.Consumer.IsConnected().Result)) |> channel.SetResult | Seek (seekData, channel) -> Log.Logger.LogDebug("{0} Seek {1}", prefix, seekData) - try - unAckedMessageTracker.Clear() - clearIncomingMessages() - if isConsumerEpochSupported then - currentConsumerEpoch <- currentConsumerEpoch + %1UL - do! resetStreams (fun consumer -> - match seekData with - | SeekType.Timestamp ts -> consumer.SeekAsync(ts) - | SeekType.MessageId msgId -> consumer.SeekAsync(msgId)) - channel.SetResult() - with Flatten ex -> - channel.SetException ex + if pendingChildren.Count > 0 then + channel.SetException(pendingChildrenException "seek") + else + try + unAckedMessageTracker.Clear() + clearIncomingMessages() + if isConsumerEpochSupported then + currentConsumerEpoch <- currentConsumerEpoch + %1UL + do! resetStreams (fun consumer -> + match seekData with + | SeekType.Timestamp ts -> consumer.SeekAsync(ts) + | SeekType.MessageId msgId -> consumer.SeekAsync(msgId)) + channel.SetResult() + with Flatten ex -> + channel.SetException ex | SeekWithResolver (resolver, channel) -> Log.Logger.LogDebug("{0} Seek with resolver", prefix) - try - unAckedMessageTracker.Clear() - clearIncomingMessages() - if isConsumerEpochSupported then - currentConsumerEpoch <- currentConsumerEpoch + %1UL - do! resetStreams (fun consumer -> consumer.SeekAsync(resolver)) - channel.SetResult() - with Flatten ex -> - channel.SetException ex + if pendingChildren.Count > 0 then + channel.SetException(pendingChildrenException "seek") + else + try + unAckedMessageTracker.Clear() + clearIncomingMessages() + if isConsumerEpochSupported then + currentConsumerEpoch <- currentConsumerEpoch + %1UL + do! resetStreams (fun consumer -> consumer.SeekAsync(resolver)) + channel.SetResult() + with Flatten ex -> + channel.SetException ex | PatternTickTime -> @@ -1015,8 +1153,11 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration Log.Logger.LogDebug("{0} PartitionTickTime", prefix) match this.ConnectionState with | Ready -> - // Check partitions changes of passed in topics, and add new topic partitions. - do! handlePartitions() + try + // Check partitions changes of passed in topics, and add new topic partitions. + do! handlePartitions() + with ex -> + Log.Logger.LogWarning(ex, "{0} PartitionTickTime failed.", prefix) | _ -> () @@ -1027,7 +1168,7 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration try let! statsTask = consumers - |> Seq.map (fun (KeyValue(_, (consumer, _))) -> consumer.GetStats()) + |> Seq.map (fun (KeyValue(_, child)) -> child.Consumer.GetStats()) |> Task.WhenAll channel.SetResult(statsTask) with Flatten ex -> @@ -1037,27 +1178,38 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration | ReconsumeLater (msg, deliverAt, channel) -> Log.Logger.LogDebug("{0} ReconsumeLater", prefix) - let consumer, _ = consumers[msg.MessageId.TopicName] - backgroundTask { - try - do! consumer.ReconsumeLaterAsync(msg, deliverAt) - unAckedMessageTracker.Remove msg.MessageId - channel.SetResult() - with Flatten ex -> - channel.SetException ex - } |> ignore + match consumers.TryGetValue msg.MessageId.TopicName with + | true, child -> + backgroundTask { + try + do! child.Consumer.ReconsumeLaterAsync(msg, deliverAt) + unAckedMessageTracker.Remove msg.MessageId + channel.SetResult() + with Flatten ex -> + channel.SetException ex + } |> ignore + | _ -> + let ex = noChildException msg.MessageId.TopicName + interceptors.OnAcknowledge(this, msg.MessageId, ex) + channel.SetException ex | ReconsumeLaterCumulative (msg, delayTime, channel) -> Log.Logger.LogDebug("{0} ReconsumeLater", prefix) - let consumer, _ = consumers[msg.MessageId.TopicName] - backgroundTask { - try - do! consumer.ReconsumeLaterCumulativeAsync(msg, delayTime) - unAckedMessageTracker.RemoveMessagesTill msg.MessageId |> ignore - with Flatten ex -> - channel.SetException ex - } |> ignore + match consumers.TryGetValue msg.MessageId.TopicName with + | true, child -> + backgroundTask { + try + do! child.Consumer.ReconsumeLaterCumulativeAsync(msg, delayTime) + unAckedMessageTracker.RemoveMessagesTill msg.MessageId |> ignore + channel.SetResult() + with Flatten ex -> + channel.SetException ex + } |> ignore + | _ -> + let ex = noChildException msg.MessageId.TopicName + interceptors.OnAcknowledgeCumulative(this, msg.MessageId, ex) + channel.SetException ex | HasMessageAvailable channel -> @@ -1066,7 +1218,7 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration try let! results = consumers - |> Seq.map (fun (KeyValue(_, (consumer, _))) -> (consumer :?> ConsumerImpl<'T>).HasMessageAvailableAsync()) + |> Seq.map (fun (KeyValue(_, child)) -> (child.Consumer :?> ConsumerImpl<'T>).HasMessageAvailableAsync()) |> Task.WhenAll results |> Array.exists id @@ -1083,9 +1235,8 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration channel.SetResult() | _ -> this.ConnectionState <- Closing - let consumerTasks = consumers |> Seq.map(fun (KeyValue(_, (consumer, _))) -> consumer.DisposeAsync().AsTask()) try - let! _ = Task.WhenAll consumerTasks + do! closeChildren() this.ConnectionState <- Closed stopConsumer() channel.SetResult() @@ -1098,16 +1249,25 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration | Unsubscribe channel -> - continueLoop <- false Log.Logger.LogDebug("{0} Unsubscribe", prefix) match this.ConnectionState with | Closing | Closed -> + continueLoop <- false channel.SetResult() + | _ when pendingChildren.Count > 0 -> + // the consumer keeps running, so the application can unsubscribe once the children are back + channel.SetException(pendingChildrenException "unsubscribe") | _ -> + continueLoop <- false this.ConnectionState <- Closing - let consumerTasks = consumers |> Seq.map(fun (KeyValue(_, (consumer, _))) -> consumer.UnsubscribeAsync()) + let consumerTasks = consumers |> Seq.map(fun (KeyValue(_, child)) -> child.Consumer.UnsubscribeAsync()) try let! _ = Task.WhenAll consumerTasks + try + // e.g. a child of a topic that discovery removed while it was being recreated + do! closeChildren() + with Flatten ex -> + Log.Logger.LogWarning(ex, "{0} could not close all child consumers properly", prefix) this.ConnectionState <- Closed Log.Logger.LogInformation("{0} unsubscribed", prefix) stopConsumer() @@ -1122,6 +1282,7 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration if t.IsFaulted then let (Flatten ex) = t.Exception Log.Logger.LogCritical(ex, "{0} mailbox failure", prefix) + terminalError <- ex this.ConnectionState <- Failed try currentMessage |> Option.iter (replyFromStoppedMailbox ex) @@ -1154,7 +1315,7 @@ type internal MultiTopicsConsumerImpl<'T> (consumerConfig: ConsumerConfiguration member internal this.Mb with get(): Channel> = mb member internal this.Consumers = - consumers.Values |> Seq.map fst |> Seq.toArray + consumers.Values |> Seq.map _.Consumer |> Seq.toArray member this.ConsumerId with get() = consumerId diff --git a/tests/IntegrationTests/Basic.fs b/tests/IntegrationTests/Basic.fs index e5ec26ae..1d69647b 100644 --- a/tests/IntegrationTests/Basic.fs +++ b/tests/IntegrationTests/Basic.fs @@ -21,11 +21,6 @@ open Pulsar.Client.IntegrationTests open Pulsar.Client.IntegrationTests.Common open FSharp.UMX -let unwrap (ex: exn) = - match ex with - | :? AggregateException as agg -> agg.GetBaseException() - | _ -> ex - let assertMailboxException (work: Task<'T>) = task { try let! _ = work.WaitAsync(TimeSpan.FromSeconds(5.0)) @@ -37,17 +32,6 @@ let assertMailboxException (work: Task<'T>) = task { | other -> failwith $"Expected the mailbox NullReferenceException but got {other.GetType().Name}: {other.Message}" } -let assertNotConnected (work: Task<'T>) = task { - try - let! _ = work.WaitAsync(TimeSpan.FromSeconds(5.0)) - failwith "Task completed successfully after mailbox failure" - with ex -> - match unwrap ex with - | :? NotConnectedException -> () - | :? TimeoutException -> failwith "Task did not fault within 5 seconds" - | other -> failwith $"Expected NotConnectedException but got {other.GetType().Name}: {other.Message}" -} - let assertAlreadyClosed (work: Task<'T>) = task { try let! _ = work.WaitAsync(TimeSpan.FromSeconds(5.0)) @@ -59,6 +43,89 @@ let assertAlreadyClosed (work: Task<'T>) = task { | other -> failwith $"Expected AlreadyClosedException but got {other.GetType().Name}: {other.Message}" } +// A null message faults a child consumer mailbox +let private crashChildMailbox (child: ConsumerImpl) = + post child.Mb (ConsumerMessage.MessageReceived(struct (Unchecked.defaultof, Unchecked.defaultof))) + +// A null poll result faults the multi-topic consumer mailbox, and the poller reply is the in-flight request +let private crashMultiTopicMailbox (consumer: MultiTopicsConsumerImpl) (pollerReply: TaskCompletionSource) = + post consumer.Mb + (MultiTopicConsumerMessage.MessageReceived(struct (CancellationToken.None, Ok Unchecked.defaultof>), pollerReply)) + +let private waitForRecreatedChild (consumer: MultiTopicsConsumerImpl) (failedChild: ConsumerImpl) childrenCount = + waitUntil "the failed child consumer is recreated" (fun () -> task { + let! connected = (consumer :> IConsumer).IsConnected() + let children = consumer.Consumers + return connected + && children.Length = childrenCount + && children |> Array.forall (fun child -> not (Object.ReferenceEquals(child, failedChild))) + }) + +// Built without PulsarClient, so a test can choose the client configuration and the lookup service +let private createMultiTopicConsumer (clientConfig: PulsarClientConfiguration) (pool: ConnectionPool) (lookup: ILookupService) + (consumerName: string) (subscriptionName: string) (topics: string[]) = + let topicNames = topics |> Array.map TopicName + let consumerConfig = + { ConsumerConfiguration.Default with + Topics = topicNames + ConsumerName = consumerName + SubscriptionName = %subscriptionName + SubscriptionType = SubscriptionType.Exclusive } + let consumerInfos = + topicNames |> Array.map (fun topicName -> + { TopicName = topicName; Schema = Schema.BYTES(); SchemaProvider = None; Metadata = { Partitions = 0 } } + : ConsumerInitInfo) + MultiTopicsConsumerImpl.InitMultiTopic(consumerConfig, clientConfig, pool, consumerInfos, lookup, + ConsumerInterceptors.Empty, ignore) + +let private closePool (pool: ConnectionPool) (lookup: ILookupService) = + { new IAsyncDisposable with + member _.DisposeAsync() = + lookup.Dispose() + ValueTask(pool.CloseAllConnections()) } + +// Holds back the broker lookup for one topic, to keep the creation of its consumer running +type private HeldLookup(inner: ILookupService) = + let mutable held: (CompleteTopicName * TaskCompletionSource * Task) option = None + member _.Hold(topic, reached, release) = held <- Some (topic, reached, release) + interface ILookupService with + member _.GetBroker topic = + match held with + | Some (heldTopic, reached, release) when heldTopic = topic -> + reached.TrySetResult() |> ignore + task { + do! release + return! inner.GetBroker topic + } + | _ -> inner.GetBroker topic + member _.GetPartitionsForTopic topic = inner.GetPartitionsForTopic topic + member _.GetPartitionedTopicMetadata topic = inner.GetPartitionedTopicMetadata topic + member _.GetTopicsUnderNamespace(ns, isPersistent) = inner.GetTopicsUnderNamespace(ns, isPersistent) + member _.GetSchema(topic, ?schema) = inner.GetSchema(topic, ?schema = schema) + member _.UpdateServiceInfo serviceInfo = inner.UpdateServiceInfo serviceInfo + member _.Dispose() = inner.Dispose() + +type private CountingInterceptor() = + let mutable consumed = 0 + let mutable disposed = 0 + let mutable failedAcks = 0 + let mutable failedCumulativeAcks = 0 + member _.Consumed = Volatile.Read(&consumed) + member _.Disposed = Volatile.Read(&disposed) + member _.FailedAcks = Volatile.Read(&failedAcks) + member _.FailedCumulativeAcks = Volatile.Read(&failedCumulativeAcks) + interface IConsumerInterceptor with + member _.Dispose() = Interlocked.Increment(&disposed) |> ignore + member _.BeforeConsume(_, message) = + Interlocked.Increment(&consumed) |> ignore + message + member _.OnAcknowledge(_, _, exn) = + if not (isNull exn) then Interlocked.Increment(&failedAcks) |> ignore + member _.OnAcknowledgeCumulative(_, _, exn) = + if not (isNull exn) then Interlocked.Increment(&failedCumulativeAcks) |> ignore + member _.OnNegativeAcksSend(_, _) = () + member _.OnAckTimeoutSend(_, _) = () + // Runs `queue` inside the mailbox loop while it is still handling the request that completes `trigger`, // so everything it posts is queued and read only after that request returns. // `trigger` must not be created with RunContinuationsAsynchronously. @@ -405,9 +472,9 @@ let tests = Log.Debug("Finished Requests posted after the consumer is closed are already closed") } - testTask "Multi-topic consumer completes requests after a child mailbox fails" { + testTask "Multi-topic consumer recreates a child whose mailbox fails" { - Log.Debug("Started Multi-topic consumer completes requests after a child mailbox fails") + Log.Debug("Started Multi-topic consumer recreates a child whose mailbox fails") let client = getClient() let topicName1 = "public/default/topic-" + Guid.NewGuid().ToString("N") let topicName2 = "public/default/topic-" + Guid.NewGuid().ToString("N") @@ -420,31 +487,118 @@ let tests = .SubscriptionType(SubscriptionType.Exclusive) .SubscribeAsync() + let multiConsumer = consumer :?> MultiTopicsConsumerImpl + let crashed = multiConsumer.Consumers |> Array.head :?> ConsumerImpl + let crashedTopic = (crashed :> IConsumer).Topic + // moves the consumer epoch on, which the recreated child has to start from + do! consumer.RedeliverUnacknowledgedMessagesAsync().WaitAsync(TimeSpan.FromSeconds(5.0)) + // queued before the child fails, and has to outlive it + let pendingReceive = consumer.ReceiveAsync() + do! Task.Delay 200 + crashChildMailbox crashed + + do! waitForRecreatedChild multiConsumer crashed 2 + Expect.isFalse "A failed child must not complete the pending receive" pendingReceive.IsCompleted + + // the recreated child delivers the topic's messages again + let! (producer : IProducer) = + client.NewProducer() + .Topic(crashedTopic) + .EnableBatching(false) + .CreateAsync() + let! _ = producer.SendAsync([| 1uy |]) + let! (message : Message) = pendingReceive.WaitAsync(TimeSpan.FromSeconds(5.0)) + Expect.equal "Pending receive should get the message from the recreated child" [| 1uy |] (message.GetValue()) + do! consumer.AcknowledgeAsync(message.MessageId).WaitAsync(TimeSpan.FromSeconds(5.0)) + do! producer.DisposeAsync().AsTask() + do! consumer.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(5.0)) + + Log.Debug("Finished Multi-topic consumer recreates a child whose mailbox fails") + } + + testTask "Non-durable multi-topic consumer stops with the cause when a child mailbox fails" { + + Log.Debug("Started Non-durable multi-topic consumer stops with the cause when a child mailbox fails") + let client = getClient() + let topicName1 = "public/default/topic-" + Guid.NewGuid().ToString("N") + let topicName2 = "public/default/topic-" + Guid.NewGuid().ToString("N") + + let! (consumer : IConsumer) = + client.NewConsumer() + .Topics([| topicName1; topicName2 |]) + .ConsumerName("failingNonDurableChild") + .SubscriptionName("test-subscription") + .SubscriptionMode(SubscriptionMode.NonDurable) + .SubscribeAsync() + let multiConsumer = consumer :?> MultiTopicsConsumerImpl let child = multiConsumer.Consumers |> Array.head :?> ConsumerImpl - // queued before the poller stops, so stopConsumer fails this waiter directly + // queued before the parent stops, so stopConsumer fails this waiter directly let pendingReceive = consumer.ReceiveAsync() do! Task.Delay 200 - post child.Mb - (ConsumerMessage.MessageReceived(struct (Unchecked.defaultof, Unchecked.defaultof))) + crashChildMailbox child - // stopConsumer fails this waiter once the parent mailbox has stopped + // a recreated non-durable child would lose its position, so the parent stops instead Expect.throwsT2 (fun () -> pendingReceive.WaitAsync(TimeSpan.FromSeconds(15.0)).Result |> ignore) |> ignore - Expect.throwsT2 (fun () -> - consumer.ReceiveAsync().WaitAsync(TimeSpan.FromSeconds(5.0)).Result |> ignore) |> ignore + do! assertNotConnectedBecauseOf typeof (consumer.ReceiveAsync()) // these still post to the stopped mailbox, so they only return if the drain replies - Expect.throwsT2 (fun () -> - consumer.GetStats().WaitAsync(TimeSpan.FromSeconds(5.0)).Result |> ignore) |> ignore - Expect.throwsT2 (fun () -> - consumer.LastDisconnectedTimestamp().WaitAsync(TimeSpan.FromSeconds(5.0)).Result |> ignore) |> ignore - Expect.throwsT2 (fun () -> - consumer.HasReachedEndOfTopic().WaitAsync(TimeSpan.FromSeconds(5.0)).Result |> ignore) |> ignore + do! assertNotConnectedBecauseOf typeof (consumer.GetStats()) + do! assertNotConnected (consumer.LastDisconnectedTimestamp()) + do! assertNotConnected (consumer.HasReachedEndOfTopic()) let! connected = consumer.IsConnected().WaitAsync(TimeSpan.FromSeconds(5.0)) Expect.isFalse "Multi-topic consumer should report disconnected after its mailbox stops" connected do! consumer.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(5.0)) - Log.Debug("Finished Multi-topic consumer completes requests after a child mailbox fails") + Log.Debug("Finished Non-durable multi-topic consumer stops with the cause when a child mailbox fails") + } + + testTask "Multi-topic consumer stops when a child stops after a failed unsubscribe" { + + Log.Debug("Started Multi-topic consumer stops when a child stops after a failed unsubscribe") + let client = getClient() + let topicName1 = "public/default/topic-" + Guid.NewGuid().ToString("N") + let topicName2 = "public/default/topic-" + Guid.NewGuid().ToString("N") + + let! (consumer : IConsumer) = + client.NewConsumer() + .Topics([| topicName1; topicName2 |]) + .ConsumerName("failedUnsubscribe") + .SubscriptionName("test-subscription") + .SubscriptionType(SubscriptionType.Exclusive) + .SubscribeAsync() + + let multiConsumer = consumer :?> MultiTopicsConsumerImpl + // unsubscribed after the other child, which succeeds + let crashed = multiConsumer.Consumers |> Array.last :?> ConsumerImpl + // nothing is produced, so this stays a receive waiter + let pendingReceive = consumer.ReceiveAsync() + do! Task.Delay 200 + let trigger = TaskCompletionSource() + let unsubscribe = TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously) + // Queued while the trigger is handled: Unsubscribe is read before any child stop is reported, and the + // crashed child has failed by the time Unsubscribe reaches it. + queueBehind trigger (fun () -> + post multiConsumer.Mb (MultiTopicConsumerMessage.Unsubscribe unsubscribe) + crashChildMailbox crashed) + post multiConsumer.Mb (MultiTopicConsumerMessage.IsConnected trigger) + + let! (unsubscribeError : exn) = task { + try + do! unsubscribe.Task.WaitAsync(TimeSpan.FromSeconds(5.0)) + return null + with ex -> + return unwrap ex + } + match unsubscribeError with + | null -> failwith "Unsubscribe should fail for the crashed child" + | :? TimeoutException -> failwith "Unsubscribe did not complete within 5 seconds" + | _ -> () + // the stopped children may have been unsubscribed, so they can't be recreated and the consumer stops + do! assertAlreadyClosed pendingReceive + do! consumer.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(5.0)) + + Log.Debug("Finished Multi-topic consumer stops when a child stops after a failed unsubscribe") } testTask "Multi-topic consumer completes the request that faults its mailbox" { @@ -457,26 +611,18 @@ let tests = let! (consumer : IConsumer) = client.NewConsumer() .Topics([| topicName1; topicName2 |]) - .ConsumerName("failingInFlightAck") + .ConsumerName("failingInFlightPoll") .SubscriptionName("test-subscription") .SubscriptionType(SubscriptionType.Exclusive) .SubscribeAsync() - // the topic lookup throws before the ack is handed to a background task - let bogusId = { MessageId.Earliest with TopicName = %"missing-topic" } - try - do! consumer.AcknowledgeAsync(bogusId).WaitAsync(TimeSpan.FromSeconds(5.0)) - failwith "Acknowledge completed successfully after faulting the mailbox" - with ex -> - match unwrap ex with - | :? System.Collections.Generic.KeyNotFoundException -> () - | :? TimeoutException -> failwith "Acknowledge did not fault within 5 seconds" - | other -> failwith $"Expected KeyNotFoundException but got {other.GetType().Name}: {other.Message}" + let pollerReply = TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously) + crashMultiTopicMailbox (consumer :?> MultiTopicsConsumerImpl) pollerReply + do! assertMailboxException pollerReply.Task - Expect.throwsT2 (fun () -> - consumer.GetStats().WaitAsync(TimeSpan.FromSeconds(5.0)).Result |> ignore) |> ignore - Expect.throwsT2 (fun () -> - consumer.LastDisconnectedTimestamp().WaitAsync(TimeSpan.FromSeconds(5.0)).Result |> ignore) |> ignore + // the drain reports why the consumer stopped + do! assertNotConnectedBecauseOf typeof (consumer.GetStats()) + do! assertNotConnectedBecauseOf typeof (consumer.LastDisconnectedTimestamp()) let! connected = consumer.IsConnected().WaitAsync(TimeSpan.FromSeconds(5.0)) Expect.isFalse "Multi-topic consumer should report disconnected after its mailbox stops" connected do! consumer.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(5.0)) @@ -484,7 +630,50 @@ let tests = Log.Debug("Finished Multi-topic consumer completes the request that faults its mailbox") } - testTask "Child mailbox failure stops a multi-topic consumer with a pending batch" { + testTask "Multi-topic consumer fails only the acknowledgement for an unknown topic" { + + Log.Debug("Started Multi-topic consumer fails only the acknowledgement for an unknown topic") + let client = getClient() + let topicName1 = "public/default/topic-" + Guid.NewGuid().ToString("N") + let topicName2 = "public/default/topic-" + Guid.NewGuid().ToString("N") + let interceptor = new CountingInterceptor() + + let! (consumer : IConsumer) = + client.NewConsumer() + .Topics([| topicName1; topicName2 |]) + .ConsumerName("unknownTopicAck") + .SubscriptionName("test-subscription") + .SubscriptionType(SubscriptionType.Exclusive) + .Intercept(interceptor) + .SubscribeAsync() + + // e.g. a message from a topic that pattern discovery has removed + let bogusId = { MessageId.Earliest with TopicName = %"missing-topic" } + do! assertNotConnected (consumer.AcknowledgeAsync(bogusId)) + do! assertNotConnected (consumer.NegativeAcknowledge(bogusId)) + do! assertNotConnected (consumer.AcknowledgeCumulativeAsync(bogusId)) + // as by a single-topic consumer, e.g. so tracing ends for the message + Expect.equal "Interceptors should be told the acknowledgement failed" 1 interceptor.FailedAcks + Expect.equal "Interceptors should be told the cumulative acknowledgement failed" 1 interceptor.FailedCumulativeAcks + let! connected = consumer.IsConnected().WaitAsync(TimeSpan.FromSeconds(5.0)) + Expect.isTrue "Multi-topic consumer should stay connected" connected + + let! (producer : IProducer) = + client.NewProducer() + .Topic(topicName1) + .EnableBatching(false) + .CreateAsync() + let! _ = producer.SendAsync([| 1uy |]) + let! (message : Message) = consumer.ReceiveAsync().WaitAsync(TimeSpan.FromSeconds(5.0)) + Expect.equal "Multi-topic consumer should keep consuming" [| 1uy |] (message.GetValue()) + do! consumer.AcknowledgeAsync(message.MessageId).WaitAsync(TimeSpan.FromSeconds(5.0)) + do! producer.DisposeAsync().AsTask() + do! consumer.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(5.0)) + + Log.Debug("Finished Multi-topic consumer fails only the acknowledgement for an unknown topic") + } + + testTask "Child mailbox failure keeps a pending batch receive alive" { let client = getClient() let topics = [| "public/default/topic-" + Guid.NewGuid().ToString("N") @@ -497,18 +686,24 @@ let tests = .BatchReceivePolicy(BatchReceivePolicy(-1, 1024L, TimeSpan.FromMinutes(1.0))) .SubscribeAsync() - let child = (consumer :?> MultiTopicsConsumerImpl).Consumers |> Array.head :?> ConsumerImpl + let multiConsumer = consumer :?> MultiTopicsConsumerImpl + let crashed = multiConsumer.Consumers |> Array.head :?> ConsumerImpl + let crashedTopic = (crashed :> IConsumer).Topic let pendingBatch = consumer.BatchReceiveAsync() do! Task.Delay 200 - post child.Mb - (ConsumerMessage.MessageReceived(struct (Unchecked.defaultof, Unchecked.defaultof))) + crashChildMailbox crashed - // Buffering the close error fills the receive queue and pauses the poller before it can fail. - do! assertAlreadyClosed pendingBatch - do! assertNotConnected (consumer.ReceiveAsync()) - do! assertNotConnected (consumer.BatchReceiveAsync()) - let! connected = consumer.IsConnected().WaitAsync(TimeSpan.FromSeconds(5.0)) - Expect.isFalse "Parent must stop before completing pending receives" connected + // A buffered close error would fill the one-message queue, so this also checks that it isn't buffered. + do! waitForRecreatedChild multiConsumer crashed 2 + Expect.isFalse "A failed child must not complete the pending batch" pendingBatch.IsCompleted + use! (producer : IProducer) = + client.NewProducer() + .Topic(crashedTopic) + .EnableBatching(false) + .CreateAsync() + let! _ = producer.SendAsync(Array.create 1024 1uy) + let! (messages : Messages) = pendingBatch.WaitAsync(TimeSpan.FromSeconds(5.0)) + Expect.equal "Pending batch should get the message from the recreated child" 1 messages.Count } testTask "Close queued behind a multi-topic consumer mailbox failure completes only after cleanup" { @@ -533,28 +728,21 @@ let tests = (MultiTopicConsumerMessage.Receive { CancellationToken = CancellationToken.None; MessageChannel = pendingReceive }) let trigger = TaskCompletionSource() - let ack = TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously) + let pollerReply = TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously) let close = TaskCompletionSource() - // queued while the trigger is being handled: the topic lookup for the ack faults the mailbox + // queued while the trigger is being handled: the null poll result faults the mailbox // and the Close is left for the drain queueBehind trigger (fun () -> - let bogusId = { MessageId.Earliest with TopicName = %"missing-topic" } - post multiConsumer.Mb (MultiTopicConsumerMessage.Acknowledge(ack, bogusId, None)) + crashMultiTopicMailbox multiConsumer pollerReply post multiConsumer.Mb (MultiTopicConsumerMessage.Close close)) let atClose = - snapshotOnCompletion close (fun () -> ack.Task.IsCompleted, pendingReceive.Task.IsCompleted) + snapshotOnCompletion close (fun () -> pollerReply.Task.IsCompleted, pendingReceive.Task.IsCompleted) post multiConsumer.Mb (MultiTopicConsumerMessage.IsConnected trigger) - let! (ackCompleted, receiveCompleted) = atClose.WaitAsync(TimeSpan.FromSeconds(5.0)) - Expect.isTrue "In-flight request should be failed before the queued Close completes" ackCompleted + let! (pollCompleted, receiveCompleted) = atClose.WaitAsync(TimeSpan.FromSeconds(5.0)) + Expect.isTrue "In-flight request should be failed before the queued Close completes" pollCompleted Expect.isTrue "Receive waiter should be failed before the queued Close completes" receiveCompleted - try - do! ack.Task.WaitAsync(TimeSpan.FromSeconds(5.0)) - failwith "Acknowledge completed successfully after faulting the mailbox" - with ex -> - match unwrap ex with - | :? System.Collections.Generic.KeyNotFoundException -> () - | other -> failwith $"Expected KeyNotFoundException but got {other.GetType().Name}: {other.Message}" + do! assertMailboxException pollerReply.Task do! assertAlreadyClosed pendingReceive.Task // a Failed transition after the drain's Close would overwrite Closed and make this NotConnectedException do! assertAlreadyClosed (consumer.GetStats()) @@ -563,9 +751,9 @@ let tests = Log.Debug("Finished Close queued behind a multi-topic consumer mailbox failure completes only after cleanup") } - testTask "Multi-topic consumer is closed at the broker when a child mailbox fails" { + testTask "Multi-topic consumer closes a recreated child at the broker" { - Log.Debug("Started Multi-topic consumer is closed at the broker when a child mailbox fails") + Log.Debug("Started Multi-topic consumer closes a recreated child at the broker") let client = getClient() let topicName1 = "public/default/topic-" + Guid.NewGuid().ToString("N") let topicName2 = "public/default/topic-" + Guid.NewGuid().ToString("N") @@ -580,23 +768,185 @@ let tests = .SubscribeAsync() let multiConsumer = consumer :?> MultiTopicsConsumerImpl - let children = multiConsumer.Consumers |> Array.map (fun c -> c :?> ConsumerImpl) - let crashed = children[0] - // the crashed child unregisters itself; the other child stays until the parent Close disposes it - let survivingTopic = - let crashedTopic = (crashed :> IConsumer).Topic - if crashedTopic.Contains(topicName1) then topicName2 else topicName1 - let pendingReceive = consumer.ReceiveAsync() - do! Task.Delay 200 - post crashed.Mb - (ConsumerMessage.MessageReceived(struct (Unchecked.defaultof, Unchecked.defaultof))) + let crashed = multiConsumer.Consumers |> Array.head :?> ConsumerImpl + crashChildMailbox crashed + // the exclusive subscription only accepts the recreated child once the crashed one is closed at the broker + do! waitForRecreatedChild multiConsumer crashed 2 + do! consumer.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(5.0)) - do! assertAlreadyClosed pendingReceive + // both the recreated and the surviving child were closed, so both topics take a new exclusive consumer + for topicName in [ topicName1; topicName2 ] do + let! (consumer2 : IConsumer) = + client.NewConsumer() + .Topic(topicName) + .ConsumerName("replacement") + .SubscriptionName(subscriptionName) + .SubscriptionType(SubscriptionType.Exclusive) + .SubscribeAsync() + .WaitAsync(TimeSpan.FromSeconds(15.0)) + do! consumer2.UnsubscribeAsync() + + Log.Debug("Finished Multi-topic consumer closes a recreated child at the broker") + } + + testTask "Closing a multi-topic consumer waits for a child that is being recreated" { + + Log.Debug("Started Closing a multi-topic consumer waits for a child that is being recreated") + let topicName1 = "public/default/topic-" + Guid.NewGuid().ToString("N") + let topicName2 = "public/default/topic-" + Guid.NewGuid().ToString("N") + let subscriptionName = "test-subscription" + let config = { PulsarClientConfiguration.Default with ServiceAddresses = [| Uri(pulsarAddress) |] } + let pool = ConnectionPool(config) + let lookup = new HeldLookup(new BinaryLookupService(config, pool)) + use _ = closePool pool lookup + + let! (multiConsumer : MultiTopicsConsumerImpl) = + createMultiTopicConsumer config pool lookup "closeDuringRecreation" subscriptionName [| topicName1; topicName2 |] + let consumer = multiConsumer :> IConsumer + let crashed = multiConsumer.Consumers |> Array.head :?> ConsumerImpl + let crashedTopic = (crashed :> IConsumer).Topic + let lookupReached = TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously) + let release = TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously) + lookup.Hold(%crashedTopic, lookupReached, release.Task) + crashChildMailbox crashed + do! lookupReached.Task.WaitAsync(TimeSpan.FromSeconds(15.0)) + + // the recreated child is still waiting for its broker + let closing = consumer.DisposeAsync().AsTask() + do! Task.Delay 500 + Expect.isFalse "Close must wait for the child that is being recreated" closing.IsCompleted + release.SetResult() + do! closing.WaitAsync(TimeSpan.FromSeconds(15.0)) + + // the recreated child was closed at the broker before Close completed, so the topic takes a new exclusive consumer + let! (consumer2 : IConsumer) = + getClient().NewConsumer() + .Topic(crashedTopic) + .ConsumerName("replacement") + .SubscriptionName(subscriptionName) + .SubscriptionType(SubscriptionType.Exclusive) + .SubscribeAsync() + .WaitAsync(TimeSpan.FromSeconds(15.0)) + do! consumer2.UnsubscribeAsync() + + Log.Debug("Finished Closing a multi-topic consumer waits for a child that is being recreated") + } + + testTask "Unsubscribing waits for a child that is still being recreated for a removed topic" { + + Log.Debug("Started Unsubscribing waits for a child that is still being recreated for a removed topic") + let topicPrefix = $"persistent://public/default/unsubscribe-recreating-{Guid.NewGuid():N}-" + let removedTopic = TopicName(topicPrefix + "removed") + let remainingTopic = TopicName(topicPrefix + "remaining") + let subscriptionName = "test-subscription" + // pattern children never create topics + for topic in [ removedTopic; remainingTopic ] do + let! (producer : IProducer) = getClient().NewProducer().Topic(topic.ToString()).CreateAsync() + do! producer.DisposeAsync().AsTask() + let config = { PulsarClientConfiguration.Default with ServiceAddresses = [| Uri(pulsarAddress) |] } + let pool = ConnectionPool(config) + let lookup = new HeldLookup(new BinaryLookupService(config, pool)) + use _ = closePool pool lookup + let consumerInfo topic : ConsumerInitInfo = + { TopicName = topic; Schema = Schema.BYTES(); SchemaProvider = None; Metadata = { Partitions = 0 } } + let patternInfo = + { InitialTopics = [| consumerInfo removedTopic; consumerInfo remainingTopic |] + GetTopics = fun () -> Task.FromResult [| remainingTopic |] + GetConsumerInfo = consumerInfo >> Task.FromResult } + let consumerConfig = + { ConsumerConfiguration.Default with + TopicsPattern = topicPrefix + ".*" + ConsumerName = "unsubscribeDuringRecreation" + SubscriptionName = %subscriptionName + SubscriptionType = SubscriptionType.Exclusive + PatternAutoDiscoveryPeriod = TimeSpan.FromHours(1.0) } + + let! (multiConsumer : MultiTopicsConsumerImpl) = + MultiTopicsConsumerImpl.InitPattern(consumerConfig, config, pool, patternInfo, lookup, + ConsumerInterceptors.Empty, ignore) + let consumer = multiConsumer :> IConsumer + let crashed = + multiConsumer.Consumers + |> Array.find (fun child -> %child.Topic = removedTopic.CompleteTopicName) :?> ConsumerImpl + let lookupReached = TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously) + let release = TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously) + lookup.Hold(removedTopic.CompleteTopicName, lookupReached, release.Task) + crashChildMailbox crashed + do! lookupReached.Task.WaitAsync(TimeSpan.FromSeconds(15.0)) + + // discovery removes the topic while its child is still being recreated, so Unsubscribe doesn't refuse + post multiConsumer.Mb MultiTopicConsumerMessage.PatternTickTime + let! connected = consumer.IsConnected().WaitAsync(TimeSpan.FromSeconds(5.0)) + Expect.isTrue "Nothing should be pending once discovery removed the topic" connected + let unsubscribing = consumer.UnsubscribeAsync() + do! Task.Delay 500 + Expect.isFalse "Unsubscribe must wait for the child that is being recreated" unsubscribing.IsCompleted + release.SetResult() + do! unsubscribing.WaitAsync(TimeSpan.FromSeconds(15.0)) + + // the child created for the removed topic was closed before Unsubscribe completed + let! (consumer2 : IConsumer) = + getClient().NewConsumer() + .Topic(removedTopic.ToString()) + .ConsumerName("replacement") + .SubscriptionName(subscriptionName) + .SubscriptionType(SubscriptionType.Exclusive) + .SubscribeAsync() + .WaitAsync(TimeSpan.FromSeconds(15.0)) + do! consumer2.UnsubscribeAsync() + + Log.Debug("Finished Unsubscribing waits for a child that is still being recreated for a removed topic") + } + + testTask "A child recreated across a redelivery is created again at the new epoch" { + + Log.Debug("Started A child recreated across a redelivery is created again at the new epoch") + let topicName1 = "public/default/topic-" + Guid.NewGuid().ToString("N") + let topicName2 = "public/default/topic-" + Guid.NewGuid().ToString("N") + let subscriptionName = "test-subscription" + let config = { PulsarClientConfiguration.Default with ServiceAddresses = [| Uri(pulsarAddress) |] } + let pool = ConnectionPool(config) + let lookup = new HeldLookup(new BinaryLookupService(config, pool)) + use _ = closePool pool lookup + + let! (multiConsumer : MultiTopicsConsumerImpl) = + createMultiTopicConsumer config pool lookup "staleRecreatedChild" subscriptionName [| topicName1; topicName2 |] + let consumer = multiConsumer :> IConsumer + let crashed = multiConsumer.Consumers |> Array.head :?> ConsumerImpl + let crashedTopic = (crashed :> IConsumer).Topic + let firstLookup = TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously) + let releaseFirst = TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously) + lookup.Hold(%crashedTopic, firstLookup, releaseFirst.Task) + crashChildMailbox crashed + do! firstLookup.Task.WaitAsync(TimeSpan.FromSeconds(15.0)) + + // the child being created subscribes at the old epoch, so its messages would be discarded + do! consumer.RedeliverUnacknowledgedMessagesAsync().WaitAsync(TimeSpan.FromSeconds(5.0)) + let secondLookup = TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously) + let releaseSecond = TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously) + lookup.Hold(%crashedTopic, secondLookup, releaseSecond.Task) + releaseFirst.SetResult() + // the stale child was discarded, and the child is being created again + do! secondLookup.Task.WaitAsync(TimeSpan.FromSeconds(15.0)) + releaseSecond.SetResult() + do! waitForRecreatedChild multiConsumer crashed 2 + + let! (producer : IProducer) = + getClient().NewProducer() + .Topic(crashedTopic) + .EnableBatching(false) + .CreateAsync() + let! _ = producer.SendAsync([| 1uy |]) + let! (message : Message) = consumer.ReceiveAsync().WaitAsync(TimeSpan.FromSeconds(5.0)) + Expect.equal "The child created at the new epoch should deliver the topic's messages" [| 1uy |] (message.GetValue()) + do! consumer.AcknowledgeAsync(message.MessageId).WaitAsync(TimeSpan.FromSeconds(5.0)) + do! producer.DisposeAsync().AsTask() do! consumer.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(5.0)) + // neither the discarded child nor its replacement is still subscribed let! (consumer2 : IConsumer) = - client.NewConsumer() - .Topic(survivingTopic) + getClient().NewConsumer() + .Topic(crashedTopic) .ConsumerName("replacement") .SubscriptionName(subscriptionName) .SubscriptionType(SubscriptionType.Exclusive) @@ -604,7 +954,93 @@ let tests = .WaitAsync(TimeSpan.FromSeconds(15.0)) do! consumer2.UnsubscribeAsync() - Log.Debug("Finished Multi-topic consumer is closed at the broker when a child mailbox fails") + Log.Debug("Finished A child recreated across a redelivery is created again at the new epoch") + } + + testTask "Multi-topic consumer refuses to seek or unsubscribe while a child is being recreated" { + + Log.Debug("Started Multi-topic consumer refuses to seek or unsubscribe while a child is being recreated") + let topicName1 = "public/default/topic-" + Guid.NewGuid().ToString("N") + let topicName2 = "public/default/topic-" + Guid.NewGuid().ToString("N") + // the failed child waits longer than the test runs before it is recreated + let config = + { PulsarClientConfiguration.Default with + ServiceAddresses = [| Uri(pulsarAddress) |] + InitialBackoffInterval = TimeSpan.FromMinutes(5.0) + MaxBackoffInterval = TimeSpan.FromMinutes(5.0) } + let pool = ConnectionPool(config) + let lookup = new BinaryLookupService(config, pool) :> ILookupService + use _ = closePool pool lookup + + let! (multiConsumer : MultiTopicsConsumerImpl) = + createMultiTopicConsumer config pool lookup "pendingChildRequests" "test-subscription" [| topicName1; topicName2 |] + let consumer = multiConsumer :> IConsumer + let crashed = multiConsumer.Consumers |> Array.head :?> ConsumerImpl + let remainingTopic = (multiConsumer.Consumers |> Array.last).Topic + crashChildMailbox crashed + do! waitUntil "the failed child consumer is detached" (fun () -> task { + let! connected = consumer.IsConnected() + return not connected && multiConsumer.Consumers.Length = 1 + }) + + do! assertNotConnected (consumer.SeekAsync(MessageId.Earliest)) + // the subscription of the failed child's topic would be left behind + do! assertNotConnected (consumer.UnsubscribeAsync()) + + // the consumer keeps running + let! (producer : IProducer) = + getClient().NewProducer() + .Topic(remainingTopic) + .EnableBatching(false) + .CreateAsync() + let! _ = producer.SendAsync([| 1uy |]) + let! (message : Message) = consumer.ReceiveAsync().WaitAsync(TimeSpan.FromSeconds(5.0)) + Expect.equal "The remaining topic should still be consumed" [| 1uy |] (message.GetValue()) + do! consumer.AcknowledgeAsync(message.MessageId).WaitAsync(TimeSpan.FromSeconds(5.0)) + do! producer.DisposeAsync().AsTask() + do! consumer.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(5.0)) + + Log.Debug("Finished Multi-topic consumer refuses to seek or unsubscribe while a child is being recreated") + } + + testTask "Recreating a multi-topic consumer's child keeps the interceptors" { + + Log.Debug("Started Recreating a multi-topic consumer's child keeps the interceptors") + let client = getClient() + let topicName1 = "public/default/topic-" + Guid.NewGuid().ToString("N") + let topicName2 = "public/default/topic-" + Guid.NewGuid().ToString("N") + let interceptor = new CountingInterceptor() + + let! (consumer : IConsumer) = + client.NewConsumer() + .Topics([| topicName1; topicName2 |]) + .ConsumerName("recreatedChildInterceptors") + .SubscriptionName("test-subscription") + .SubscriptionType(SubscriptionType.Exclusive) + .Intercept(interceptor) + .SubscribeAsync() + + let multiConsumer = consumer :?> MultiTopicsConsumerImpl + let crashed = multiConsumer.Consumers |> Array.head :?> ConsumerImpl + let crashedTopic = (crashed :> IConsumer).Topic + crashChildMailbox crashed + do! waitForRecreatedChild multiConsumer crashed 2 + Expect.equal "A stopped child must not dispose the interceptors the parent still uses" 0 interceptor.Disposed + + let! (producer : IProducer) = + client.NewProducer() + .Topic(crashedTopic) + .EnableBatching(false) + .CreateAsync() + let! _ = producer.SendAsync([| 1uy |]) + let! (message : Message) = consumer.ReceiveAsync().WaitAsync(TimeSpan.FromSeconds(5.0)) + Expect.equal "The recreated child should run the interceptors" 1 interceptor.Consumed + do! consumer.AcknowledgeAsync(message.MessageId).WaitAsync(TimeSpan.FromSeconds(5.0)) + do! producer.DisposeAsync().AsTask() + do! consumer.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(5.0)) + Expect.equal "Closing the parent should dispose the interceptors once" 1 interceptor.Disposed + + Log.Debug("Finished Recreating a multi-topic consumer's child keeps the interceptors") } testTask "Requests posted after the multi-topic consumer is closed are already closed" { diff --git a/tests/IntegrationTests/Common.fs b/tests/IntegrationTests/Common.fs index 9b3ce6b2..ffb36e7b 100644 --- a/tests/IntegrationTests/Common.fs +++ b/tests/IntegrationTests/Common.fs @@ -262,3 +262,59 @@ let consumeAndVerifyMessages (consumer: IConsumer) consumerName (expecte if expectedMessages |> Array.contains received |> not then failwith <| sprintf "Received unexpected message '%s' consumer %s" received consumerName } + +let unwrap (ex: exn) = + match ex with + | :? AggregateException as agg -> agg.GetBaseException() + | _ -> ex + +let assertNotConnected (work: Task<'T>) = task { + try + let! _ = work.WaitAsync(TimeSpan.FromSeconds(5.0)) + failwith "Task completed successfully after mailbox failure" + with ex -> + match unwrap ex with + | :? NotConnectedException -> () + | :? System.TimeoutException -> failwith "Task did not fault within 5 seconds" + | other -> failwith $"Expected NotConnectedException but got {other.GetType().Name}: {other.Message}" +} + +// The failure has to keep its cause, so the innermost exception is the one that stopped the consumer. +let assertNotConnectedBecauseOf (cause: Type) (work: Task<'T>) = task { + let mutable error = null + try + let! _ = work.WaitAsync(TimeSpan.FromSeconds(5.0)) + () + with ex -> + error <- ex + match error with + | null -> failwith "Task completed successfully after mailbox failure" + | ex -> + match unwrap ex with + | :? NotConnectedException as notConnected when cause.IsInstanceOfType(notConnected.GetBaseException()) -> () + | :? NotConnectedException as notConnected -> + failwith $"Expected the failure to be caused by {cause.Name} but got {notConnected.GetBaseException()}" + | :? System.TimeoutException -> failwith "Task did not fault within 5 seconds" + | other -> failwith $"Expected NotConnectedException but got {other.GetType().Name}: {other.Message}" +} + +// Retries until the condition holds. Internal state read from the test thread can change underneath it, +// so a failing read is retried too. A condition that doesn't complete before the deadline counts as false. +let waitUntil (description: string) (condition: unit -> Task) = task { + let deadline = DateTime.UtcNow.AddSeconds 15.0 + let mutable satisfied = false + while not satisfied do + let remaining = max TimeSpan.Zero (deadline - DateTime.UtcNow) + let! result = task { + try + return! condition().WaitAsync(remaining) + with _ -> + return false + } + if result then + satisfied <- true + elif DateTime.UtcNow >= deadline then + failwith $"Timed out waiting until {description}" + else + do! Task.Delay 100 +} diff --git a/tests/IntegrationTests/MultiTopic.fs b/tests/IntegrationTests/MultiTopic.fs index 0c4fdba2..7a42c28c 100644 --- a/tests/IntegrationTests/MultiTopic.fs +++ b/tests/IntegrationTests/MultiTopic.fs @@ -196,6 +196,110 @@ let tests = Expect.isTrue stillConnected "Intentional child removal must not fail the parent poller" } + testTask "Acknowledging a message from a removed pattern topic fails only that request" { + let timeout = TimeSpan.FromSeconds(5.0) + let topicPrefix = $"persistent://public/default/pattern-removed-ack-{Guid.NewGuid():N}-" + let removedTopic = TopicName(topicPrefix + "removed") + let remainingTopic = TopicName(topicPrefix + "remaining") + let client = getClient() + use! (removedProducer: IProducer) = + client.NewProducer().Topic(removedTopic.ToString()).EnableBatching(false).CreateAsync() + use! (remainingProducer: IProducer) = + client.NewProducer().Topic(remainingTopic.ToString()).EnableBatching(false).CreateAsync() + + let config = { PulsarClientConfiguration.Default with ServiceAddresses = [| Uri(pulsarAddress) |] } + let pool = ConnectionPool(config) + let lookup = new BinaryLookupService(config, pool) :> ILookupService + use poolCleanup = + { new IAsyncDisposable with + member _.DisposeAsync() = + lookup.Dispose() + ValueTask(pool.CloseAllConnections()) } + let consumerInfo topic : ConsumerInitInfo = + { TopicName = topic; Schema = Schema.BYTES(); SchemaProvider = None; Metadata = { Partitions = 0 } } + let patternInfo = + { InitialTopics = [| consumerInfo removedTopic; consumerInfo remainingTopic |] + GetTopics = fun () -> Task.FromResult [| remainingTopic |] + GetConsumerInfo = consumerInfo >> Task.FromResult } + let consumerConfig = + { ConsumerConfiguration.Default with + TopicsPattern = topicPrefix + ".*" + SubscriptionName = %"pattern-removed-ack" + PatternAutoDiscoveryPeriod = TimeSpan.FromHours(1.0) } + use! (consumer: MultiTopicsConsumerImpl) = + MultiTopicsConsumerImpl.InitPattern(consumerConfig, config, pool, patternInfo, lookup, + ConsumerInterceptors.Empty, ignore) + let publicConsumer = consumer :> IConsumer + + let! _ = removedProducer.SendAsync [| 1uy |] + let! (removedMessage: Message) = publicConsumer.ReceiveAsync().WaitAsync(timeout) + post consumer.Mb MultiTopicConsumerMessage.PatternTickTime + // answered after the discovery has been handled + let! _ = publicConsumer.IsConnected().WaitAsync(timeout) + Expect.equal consumer.Consumers.Length 1 "Only the discovered child should remain" + + // the application still holds a message from the removed topic + do! assertNotConnected (publicConsumer.AcknowledgeAsync(removedMessage.MessageId)) + do! assertNotConnected (publicConsumer.NegativeAcknowledge(removedMessage.MessageId)) + let! connected = publicConsumer.IsConnected().WaitAsync(timeout) + Expect.isTrue connected "Acknowledging a removed topic's message must not fail the parent" + + let! _ = remainingProducer.SendAsync [| 2uy |] + let! (message: Message) = publicConsumer.ReceiveAsync().WaitAsync(timeout) + Expect.equal message.Data [| 2uy |] "The remaining topic should still be consumed" + do! publicConsumer.AcknowledgeAsync(message.MessageId).WaitAsync(timeout) + } + + testTask "Pattern consumer keeps running after a matched topic is deleted" { + let timeout = TimeSpan.FromSeconds(10.0) + let topicPrefix = $"public/default/pattern-deleted-{Guid.NewGuid():N}-" + let deletedTopic = topicPrefix + "deleted" + let remainingTopic = topicPrefix + "remaining" + let client = getClient() + let! (deletedProducer: IProducer) = + client.NewProducer().Topic(deletedTopic).EnableBatching(false).CreateAsync() + use! (remainingProducer: IProducer) = + client.NewProducer().Topic(remainingTopic).EnableBatching(false).CreateAsync() + use! (consumer: IConsumer) = + client.NewConsumer() + .TopicsPattern(topicPrefix + ".*") + .SubscriptionName("pattern-deleted") + .PatternAutoDiscoveryPeriod(TimeSpan.FromHours(1.0)) + .SubscribeAsync() + let multiConsumer = consumer :?> MultiTopicsConsumerImpl + Expect.equal multiConsumer.Consumers.Length 2 "Both topics should be subscribed" + do! deletedProducer.DisposeAsync().AsTask() + + // pattern children never create topics, so the deleted topic's child can't subscribe again and stops + let! (response: Net.Http.HttpResponseMessage) = commonHttpClient.DeleteAsync($"{pulsarHttpAddress}/admin/v2/persistent/{deletedTopic}?force=true") + response.EnsureSuccessStatusCode() |> ignore + do! waitUntil "the deleted topic's child is detached" (fun () -> Task.FromResult(multiConsumer.Consumers.Length = 1)) + + // restarting every child stream used to reach the stopped child and fail the whole consumer + do! consumer.RedeliverUnacknowledgedMessagesAsync().WaitAsync(timeout) + // the topic stays pending until discovery removes it, so these would skip it + let! connected = consumer.IsConnected().WaitAsync(timeout) + Expect.isFalse connected "The deleted topic's child is not consuming" + do! assertNotConnected (consumer.SeekAsync(MessageId.Earliest)) + do! assertNotConnected (consumer.UnsubscribeAsync()) + let! _ = remainingProducer.SendAsync [| 1uy |] + let! (message: Message) = consumer.ReceiveAsync().WaitAsync(timeout) + Expect.equal message.Data [| 1uy |] "The remaining topic should still be consumed" + do! consumer.AcknowledgeAsync(message.MessageId).WaitAsync(timeout) + + // once the topic exists again, the pending child is recreated without waiting for discovery + use! (recreatedProducer: IProducer) = + client.NewProducer().Topic(deletedTopic).EnableBatching(false).CreateAsync() + do! waitUntil "the recreated topic is subscribed again" (fun () -> task { + let! connected = consumer.IsConnected() + return connected && multiConsumer.Consumers.Length = 2 + }) + let! _ = recreatedProducer.SendAsync [| 2uy |] + let! (message: Message) = consumer.ReceiveAsync().WaitAsync(timeout) + Expect.equal message.Data [| 2uy |] "The recreated topic should be consumed" + do! consumer.AcknowledgeAsync(message.MessageId).WaitAsync(timeout) + } + ptestTask "Eternal loop to test cluster modifications removal/addition" { Log.Debug("Started Eternal loop to test cluster modifications removal/addition")