Replicated consumer reconnect logic in producer #238

fsprojects · Jul 31, 2024 · f55aff4 · f55aff4
1 parent aebde9c
commit f55aff4
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 6 deletions.
diff --git a/src/Pulsar.Client/Internal/ProducerImpl.fs b/src/Pulsar.Client/Internal/ProducerImpl.fs
@@ -580,11 +580,14 @@ type internal ProducerImpl<'T> private (producerConfig: ProducerConfiguration, c
                 connectionHandler.ConnectionClosed clientCnx
                 clientCnx.RemoveProducer(producerId)
 
-            | ProducerMessage.ConnectionFailed _ ->
+            | ProducerMessage.ConnectionFailed ex ->
 
                 Log.Logger.LogDebug("{0} ConnectionFailed", prefix)
-                if Stopwatch.GetElapsedTime(createProducerStartTime) > clientConfig.OperationTimeout then
-                    Log.Logger.LogInformation("{0} creation failed", prefix)
+                let nonRetriableError = ex |> PulsarClientException.isRetriableError |> not
+                let timeout = Stopwatch.GetElapsedTime(createProducerStartTime) > clientConfig.OperationTimeout
+                if ((nonRetriableError || timeout) && producerCreatedTsc.TrySetException(ex)) then
+                    Log.Logger.LogInformation("{0} creation failed {1}", prefix,
+                                                if nonRetriableError then "with unretriableError" else "after timeout")
                     connectionHandler.Failed()
                     stopProducer()
                     continueLoop <- false

diff --git a/src/Pulsar.Client/Pulsar.Client.fsproj b/src/Pulsar.Client/Pulsar.Client.fsproj
@@ -7,17 +7,17 @@
     <Title>Pulsar.Client</Title>
     <RootNamespace>Pulsar.Client</RootNamespace>
     <AssemblyName>Pulsar.Client</AssemblyName>
-    <Version>3.5.2</Version>
+    <Version>3.5.3</Version>
     <Company>F# community</Company>
     <Description>.NET client library for Apache Pulsar</Description>
     <RepositoryUrl>https://github.com/fsprojects/pulsar-client-dotnet</RepositoryUrl>
-    <PackageReleaseNotes>Fix for missing key in reconsumed message</PackageReleaseNotes>
+    <PackageReleaseNotes>Replicated consumer reconnect logic in producer</PackageReleaseNotes>
     <PackageLicenseExpression>MIT</PackageLicenseExpression>
     <PackageProjectUrl>https://github.com/fsprojects/pulsar-client-dotnet</PackageProjectUrl>
     <RepositoryType>git</RepositoryType>
     <PackageTags>pulsar</PackageTags>
     <Authors>F# community</Authors>
-    <PackageVersion>3.5.2</PackageVersion>
+    <PackageVersion>3.5.3</PackageVersion>
     <DebugType>portable</DebugType>
     <GenerateDocumentationFile>true</GenerateDocumentationFile>
     <PackageReadmeFile>README.md</PackageReadmeFile>