You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by bl...@apache.org on 2025/03/17 13:25:29 UTC
(pulsar-dotpulsar) branch master updated: The consumer no longer sends the schema to the broker when the schema type is 'None'. Make ready for release 4.2.2
This is an automated email from the ASF dual-hosted git repository.
blankensteiner pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar-dotpulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 01df3b0 The consumer no longer sends the schema to the broker when the schema type is 'None'. Make ready for release 4.2.2
01df3b0 is described below
commit 01df3b0b5b17eef820109d500b29d0978035abd0
Author: Daniel Blankensteiner <db...@vmail.dk>
AuthorDate: Mon Mar 17 14:25:21 2025 +0100
The consumer no longer sends the schema to the broker when the schema type is 'None'.
Make ready for release 4.2.2
---
CHANGELOG.md | 6 ++++++
benchmarks/Compression/Compression.csproj | 2 +-
src/DotPulsar/Internal/Consumer.cs | 6 ++++--
tests/DotPulsar.Tests/Internal/ConsumerTests.cs | 19 +++++++++++++++++++
4 files changed, 30 insertions(+), 3 deletions(-)
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 320e268..a7018b4 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,12 @@ All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+## [4.2.2] - 2025-03-17
+
+### Changed
+
+- The consumer no longer sends the schema to the broker when the schema type is 'None'
+
## [4.2.1] - 2025-03-14
### Changed
diff --git a/benchmarks/Compression/Compression.csproj b/benchmarks/Compression/Compression.csproj
index 2fb1193..bbb53c4 100644
--- a/benchmarks/Compression/Compression.csproj
+++ b/benchmarks/Compression/Compression.csproj
@@ -9,7 +9,7 @@
<ItemGroup>
<PackageReference Include="BenchmarkDotNet" Version="0.14.0" />
- <PackageReference Include="Google.Protobuf" Version="3.30.0" />
+ <PackageReference Include="Google.Protobuf" Version="3.30.1" />
<PackageReference Include="Grpc.Tools" Version="2.71.0">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
diff --git a/src/DotPulsar/Internal/Consumer.cs b/src/DotPulsar/Internal/Consumer.cs
index 4a0eec4..54b1d4c 100644
--- a/src/DotPulsar/Internal/Consumer.cs
+++ b/src/DotPulsar/Internal/Consumer.cs
@@ -434,10 +434,12 @@ public sealed class Consumer<TMessage> : IConsumer<TMessage>
ReplicateSubscriptionState = _consumerOptions.ReplicateSubscriptionState,
Subscription = _consumerOptions.SubscriptionName,
Topic = topic,
- Type = (CommandSubscribe.SubType) _consumerOptions.SubscriptionType,
- Schema = _consumerOptions.Schema.SchemaInfo.PulsarSchema
+ Type = (CommandSubscribe.SubType) _consumerOptions.SubscriptionType
};
+ if (_consumerOptions.Schema.SchemaInfo.Type != SchemaType.None)
+ subscribe.Schema = _consumerOptions.Schema.SchemaInfo.PulsarSchema;
+
foreach (var property in _consumerOptions.SubscriptionProperties)
{
var keyValue = new KeyValue { Key = property.Key, Value = property.Value };
diff --git a/tests/DotPulsar.Tests/Internal/ConsumerTests.cs b/tests/DotPulsar.Tests/Internal/ConsumerTests.cs
index d64612f..3f60b04 100644
--- a/tests/DotPulsar.Tests/Internal/ConsumerTests.cs
+++ b/tests/DotPulsar.Tests/Internal/ConsumerTests.cs
@@ -19,6 +19,7 @@ using DotPulsar.Abstractions;
using DotPulsar.Exceptions;
using DotPulsar.Extensions;
using DotPulsar.Tests.Schemas.TestSamples.AvroModels;
+using System.Text;
using System.Text.RegularExpressions;
using Xunit.Abstractions;
@@ -349,6 +350,24 @@ public sealed class ConsumerTests : IDisposable
actual["Age"].ShouldBe(expected.Age);
}
+ [Fact]
+ public async Task Subscription_WhenSubscribingToAnExistingTopicWithNoSchema_ShouldNotSubscribeWithSchemaTypeNone()
+ {
+ //Arrange
+ var topicName = await _fixture.CreateTopic(_cts.Token);
+ await using var client = CreateClient();
+ await using var producer = CreateProducer(client, topicName, Schema.ByteSequence);
+ await producer.Send(Encoding.UTF8.GetBytes("Test"), _cts.Token);
+ await using var consumer = CreateConsumer(client, topicName, Schema.ByteSequence);
+ var receiveTask = consumer.Receive(_cts.Token);
+
+ //Act
+ var exception = await Record.ExceptionAsync(receiveTask.AsTask);
+
+ //Assert
+ exception.ShouldBeNull();
+ }
+
[Fact]
public async Task Connectivity_WhenInitiallyConnectedWithNoMessagesThenGoesDown_ShouldBeAbleToReceiveWhenUpAgain()
{