Fix producer max pending messages validation inconsistency (#284)

* Fix producer max pending messages validation inconsistency

ProducerBuilder.Create validated
  MaxPendingMessages, but PulsarClient.CreateProducer accepted 0 through ProducerOptions.

  This allowed an invalid producer configuration to reach the internal send queue, making the behavior inconsistent
  depending on which API was used to create the producer.

  Add the missing validation to PulsarClient.CreateProducer and add unit tests for both creation paths.

* moved producerbuildertest.cs to the internal namespace

* moved the validation test to pulsarclienttests

Author: AmiradelBeyg

src/DotPulsar/PulsarClient.cs

@@ -60,6 +60,9 @@ public IProducer<TMessage> CreateProducer<TMessage>(ProducerOptions<TMessage> op
     {
         ThrowIfDisposed();
 
+        if (options.MaxPendingMessages == 0)
+            throw new ConfigurationException("ProducerOptions.MaxPendingMessages must be greater than 0");
+
         ICompressorFactory? compressorFactory = null;
         if (options.CompressionType != CompressionType.None)
         {

tests/DotPulsar.Tests/Internal/ProducerBuilderTests.cs

@@ -0,0 +1,47 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+namespace DotPulsar.Tests.Internal;
+
+using DotPulsar.Abstractions;
+using DotPulsar.Exceptions;
+using DotPulsar.Extensions;
+
+[Trait("Category", "Unit")]
+public sealed class ProducerBuilderTests
+{
+    [Fact]
+    public async Task Create_GivenMaxPendingMessagesIsZero_ShouldThrowConfigurationException()
+    {
+        // Arrange
+        await using var client = CreateClient();
+
+        // Act
+        var act = () => client
+            .NewProducer(Schema.String)
+            .Topic("persistent://public/default/my-topic")
+            .MaxPendingMessages(0)
+            .Create();
+
+        // Assert
+        var exception = act.ShouldThrow<ConfigurationException>();
+        exception.Message.ShouldBe("ProducerOptions.MaxPendingMessages must be greater than 0");
+    }
+
+    private static IPulsarClient CreateClient()
+        => PulsarClient
+            .Builder()
+            .ServiceUrl(new Uri("pulsar://localhost:6650"))
+            .Build();
+}

tests/DotPulsar.Tests/PulsarClientTests.cs

@@ -124,6 +124,28 @@ public async Task TokenSupplier_WhenTokenSupplierReturnValidToken_ShouldStayConn
         state.ShouldBe(ProducerState.Connected);
     }
 
+    [Fact]
+    public async Task CreateProducer_GivenMaxPendingMessagesIsZero_ShouldThrowConfigurationException()
+    {
+        // Arrange
+        await using var client = PulsarClient
+            .Builder()
+            .ServiceUrl(new Uri("pulsar://localhost:6650"))
+            .Build();
+
+        var options = new ProducerOptions<string>("persistent://public/default/my-topic", Schema.String)
+        {
+            MaxPendingMessages = 0
+        };
+
+        // Act
+        var act = () => client.CreateProducer(options);
+
+        // Assert
+        var exception = act.ShouldThrow<ConfigurationException>();
+        exception.Message.ShouldBe("ProducerOptions.MaxPendingMessages must be greater than 0");
+    }
+
     private IPulsarClient CreateClient(Func<CancellationToken, ValueTask<string>> tokenSupplier)
         => PulsarClient
         .Builder()