Expected behavior for AutoUpdatePartitions

[ppilev]

hello,
I have a question about how AutoUpdatePartitions(true) should behave when partitioned topic scales up?
we use KeyShared subscription and encountered an issue that there are no consumers attached to those new partitions thus message backlog starts building up for those partitions.
is there anything else in terms of either server or client configuration that should be considered in such scenarios?

ps: as a workaround atm we use TopicPatterns method

regards,

[Lanayx]

Hi!
This setting is on by default and consumers for new partitions should be added once new partitions are detected. If something doesn't work as you expected, you need to collect logs in debug mode, to understand what's going on.

Here is a relevant piece of code to look at:

pulsar-client-dotnet/src/Pulsar.Client/Internal/MultiTopicsConsumerImpl.fs

Lines 439 to 522 in 97f0128

	let handlePartitions() =
	backgroundTask {
	let! newPartitionsOption = getAllPartitions()
	match newPartitionsOption with
	| Some newPartitions ->
	let oldConsumersCount = consumers.Count
	let mutable totalConsumersCount = oldConsumersCount
	let topicsToUpdate =
	newPartitions
	|> Array.filter(fun (topic, partitionedTopicNames) ->
	let consumerInitInfo = partitionedTopics[topic]
	let oldPartitionsCount = consumerInitInfo.Metadata.Partitions
	let newPartitionsCount = partitionedTopicNames.Length
	if (oldPartitionsCount < newPartitionsCount) then
	Log.Logger.LogDebug("{0} partitions number. old: {1}, new: {2}, topic {3}",
	prefix, oldPartitionsCount, newPartitionsCount, topic)
	true
	elif (oldPartitionsCount > newPartitionsCount) then
	Log.Logger.LogError("{0} not support shrink topic partitions. old: {1}, new: {2}, topic: {3}",
	prefix, oldPartitionsCount, newPartitionsCount, topic)
	false
	else
	false)
	if topicsToUpdate.Length > 0 then
	Log.Logger.LogInformation("{0} adding subscription to {1} new partitions", prefix, topicsToUpdate.Length)
	let receiverQueueSize = Math.Min(consumerConfig.ReceiverQueueSize, consumerConfig.MaxTotalReceiverQueueSizeAcrossPartitions / totalConsumersCount)
	let newConsumerTasks =
	seq {
	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 = seq { partitionedTopic } |> Seq.cache }
	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
	}
	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)
	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
	Log.Logger.LogInformation("{0} disposed partially created consumers", prefix)
	| None ->
	()
	}