diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2dd7a71..1247972 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,16 +4,15 @@
 
 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).
 
-## [Unreleased]
+## [3.0.1] - 2023-09-15
 
 ### Changed
 
-- When calling GetLastMessageId(s) on a Reader or Consumer, it returns a MessageId without the topic field if
-  MessageId.Earliest is found.
+- When calling GetLastMessageId(s) on a Reader or Consumer, it returns a MessageId without the topic field if MessageId.Earliest is found
 
 ### Fixed
 
-- Fixed issue with DotPulsar client not handling connection faults for consumers and readers.
+- Fixed issue with the DotPulsar client not handling connection faults for consumers and readers
 
 ## [3.0.0] - 2023-08-30
 
diff --git a/src/DotPulsar/DotPulsar.csproj b/src/DotPulsar/DotPulsar.csproj
index 36c6bb4..0e8fbd3 100644
--- a/src/DotPulsar/DotPulsar.csproj
+++ b/src/DotPulsar/DotPulsar.csproj
@@ -2,7 +2,7 @@
 
   <PropertyGroup>
     <TargetFrameworks>netstandard2.0;netstandard2.1;net6.0;net7.0</TargetFrameworks>
-    <Version>3.0.0</Version>
+    <Version>3.0.1</Version>
     <AssemblyVersion>$(Version)</AssemblyVersion>
     <FileVersion>$(Version)</FileVersion>
     <Authors>ApachePulsar,DanskeCommodities,dblank</Authors>
diff --git a/tests/DotPulsar.Tests/ConsumerTests.cs b/tests/DotPulsar.Tests/ConsumerTests.cs
index 4c63a51..6460dbe 100644
--- a/tests/DotPulsar.Tests/ConsumerTests.cs
+++ b/tests/DotPulsar.Tests/ConsumerTests.cs
@@ -1,4 +1,4 @@
-﻿/*
+/*
  * 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
@@ -208,26 +208,15 @@
     {
         //Arrange
         var semaphoreSlim = new SemaphoreSlim(1);
-        await using var
-            client = PulsarClient.Builder().ExceptionHandler(context =>
-                {
-                    semaphoreSlim.WaitAsync();
-                    context.Result = FaultAction.Rethrow;
-                    context.ExceptionHandled = true;
-                })
-                .ServiceUrl(new Uri("pulsar://localhost:9512")) //point to a cluster that does not exists.
-                .Build();
+        await using var client = PulsarClient.Builder().ExceptionHandler(context =>
+        {
+            semaphoreSlim.WaitAsync();
+            context.Result = FaultAction.Rethrow;
+            context.ExceptionHandled = true;
+        })
+        .ServiceUrl(new Uri("pulsar://nosuchhost")).Build();
 
-        await using var consumer = client.NewConsumer(Schema.String)
-            .StateChangedHandler(changed =>
-            {
-                var topic = changed.Consumer.Topic;
-                var state = changed.ConsumerState;
-                _testOutputHelper.WriteLine($"The consumer for topic '{topic}' changed state to '{state}'");
-            })
-            .SubscriptionName("MySubscription")
-            .Topic("persistent://public/default/mytopic")
-            .Create();
+        await using var consumer = CreateConsumer(client, SubscriptionInitialPosition.Earliest, "persistent://a/b/c", "cn", "sn");
 
         var receiveTask = consumer.Receive().AsTask();
         semaphoreSlim.Release();
@@ -243,29 +232,16 @@
     public async Task Receive_WhenFaultedBeforeInvokingReceive_ShouldThrowConsumerFaultedException()
     {
         //Arrange
-        var cts = new CancellationTokenSource();
+        await using var client = PulsarClient.Builder().ExceptionHandler(context =>
+        {
+            context.Result = FaultAction.Rethrow;
+            context.ExceptionHandled = true;
+        })
+        .ServiceUrl(new Uri("pulsar://nosuchhost")).Build();
 
-        await using var
-            client = PulsarClient.Builder().ExceptionHandler(context =>
-                {
-                    context.Result = FaultAction.Rethrow;
-                    context.ExceptionHandled = true;
-                })
-                .ServiceUrl(new Uri("pulsar://localhost:9512")) //point to a cluster that does not exists.
-                .Build();
+        await using var consumer = CreateConsumer(client, SubscriptionInitialPosition.Earliest, "persistent://a/b/c", "cn", "sn");
 
-        await using var consumer = client.NewConsumer(Schema.String)
-            .StateChangedHandler(changed =>
-            {
-                var topic = changed.Consumer.Topic;
-                var state = changed.ConsumerState;
-                _testOutputHelper.WriteLine($"The consumer for topic '{topic}' changed state to '{state}'");
-            })
-            .SubscriptionName("MySubscription")
-            .Topic("persistent://public/default/mytopic")
-            .Create();
-
-        await consumer.OnStateChangeTo(ConsumerState.Faulted, cts.Token);
+        await consumer.OnStateChangeTo(ConsumerState.Faulted);
 
         //Act
         var exception = await Record.ExceptionAsync(() => consumer.Receive().AsTask());
@@ -305,18 +281,30 @@
         return messageIds;
     }
 
-    private IProducer<string> CreateProducer(IPulsarClient pulsarClient, string topicName) => pulsarClient.NewProducer(Schema.String)
+    private void LogState(ConsumerStateChanged stateChange)
+        => _testOutputHelper.WriteLine($"The consumer for topic '{stateChange.Consumer.Topic}' changed state to '{stateChange.ConsumerState}'");
+
+    private void LogState(ProducerStateChanged stateChange)
+        => _testOutputHelper.WriteLine($"The producer for topic '{stateChange.Producer.Topic}' changed state to '{stateChange.ProducerState}'");
+
+    private IProducer<string> CreateProducer(IPulsarClient pulsarClient, string topicName)
+        => pulsarClient.NewProducer(Schema.String)
         .Topic(topicName)
+        .StateChangedHandler(LogState)
         .Create();
 
-    private IConsumer<string> CreateConsumer(IPulsarClient pulsarClient, SubscriptionInitialPosition subscriptionInitialPosition,
+    private IConsumer<string> CreateConsumer(
+        IPulsarClient pulsarClient,
+        SubscriptionInitialPosition subscriptionInitialPosition,
         string topicName,
         string consumerName,
-        string subscriptionName) => pulsarClient.NewConsumer(Schema.String)
+        string subscriptionName)
+        => pulsarClient.NewConsumer(Schema.String)
         .ConsumerName(consumerName)
         .InitialPosition(subscriptionInitialPosition)
         .SubscriptionName(subscriptionName)
         .Topic(topicName)
+        .StateChangedHandler(LogState)
         .Create();
 
     private IPulsarClient CreateClient()
diff --git a/tests/DotPulsar.Tests/ReaderTests.cs b/tests/DotPulsar.Tests/ReaderTests.cs
index 6a595d6..a2c546f 100644
--- a/tests/DotPulsar.Tests/ReaderTests.cs
+++ b/tests/DotPulsar.Tests/ReaderTests.cs
@@ -1,4 +1,4 @@
-﻿/*
+/*
  * 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
@@ -210,26 +210,15 @@
     {
         //Arrange
         var semaphoreSlim = new SemaphoreSlim(1);
-        await using var
-            client = PulsarClient.Builder().ExceptionHandler(context =>
-                {
-                    semaphoreSlim.WaitAsync();
-                    context.Result = FaultAction.Rethrow;
-                    context.ExceptionHandled = true;
-                })
-                .ServiceUrl(new Uri("pulsar://localhost:9512")) //point to a cluster that does not exists.
-                .Build();
+        await using var client = PulsarClient.Builder().ExceptionHandler(context =>
+        {
+            semaphoreSlim.WaitAsync();
+            context.Result = FaultAction.Rethrow;
+            context.ExceptionHandled = true;
+        })
+        .ServiceUrl(new Uri("pulsar://nosuchhost")).Build();
 
-        await using var reader = client.NewReader(Schema.String)
-            .StartMessageId(MessageId.Earliest)
-            .StateChangedHandler(changed =>
-            {
-                var topic = changed.Reader.Topic;
-                var state = changed.ReaderState;
-                _testOutputHelper.WriteLine($"The consumer for topic '{topic}' changed state to '{state}'");
-            })
-            .Topic("persistent://public/default/mytopic")
-            .Create();
+        await using var reader = CreateReader(client, MessageId.Earliest, "persistent://a/b/c");
 
         var receiveTask = reader.Receive().AsTask();
         semaphoreSlim.Release();
@@ -245,29 +234,16 @@
     public async Task Receive_WhenFaultedBeforeInvokingReceive_ShouldThrowConsumerFaultedException()
     {
         //Arrange
-        var cts = new CancellationTokenSource();
+        await using var client = PulsarClient.Builder().ExceptionHandler(context =>
+        {
+            context.Result = FaultAction.Rethrow;
+            context.ExceptionHandled = true;
+        })
+        .ServiceUrl(new Uri("pulsar://nosuchhost")).Build();
 
-        await using var
-            client = PulsarClient.Builder().ExceptionHandler(context =>
-                {
-                    context.Result = FaultAction.Rethrow;
-                    context.ExceptionHandled = true;
-                })
-                .ServiceUrl(new Uri("pulsar://localhost:9512")) //point to a cluster that does not exists.
-                .Build();
+        await using var reader = CreateReader(client, MessageId.Earliest, "persistent://a/b/c");
 
-        await using var reader = client.NewReader(Schema.String)
-            .StartMessageId(MessageId.Earliest)
-            .StateChangedHandler(changed =>
-            {
-                var topic = changed.Reader.Topic;
-                var state = changed.ReaderState;
-                _testOutputHelper.WriteLine($"The reader for topic '{topic}' changed state to '{state}'");
-            })
-            .Topic("persistent://public/default/mytopic")
-            .Create();
-
-        await reader.OnStateChangeTo(ReaderState.Faulted, cts.Token);
+        await reader.OnStateChangeTo(ReaderState.Faulted);
 
         //Act
         var exception = await Record.ExceptionAsync(() => reader.Receive().AsTask());
@@ -276,13 +252,21 @@
         exception.Should().BeOfType<ReaderFaultedException>();
     }
 
+    private void LogState(ReaderStateChanged stateChange)
+        => _testOutputHelper.WriteLine($"The reader for topic '{stateChange.Reader.Topic}' changed state to '{stateChange.ReaderState}'");
+
+    private void LogState(ProducerStateChanged stateChange)
+        => _testOutputHelper.WriteLine($"The producer for topic '{stateChange.Producer.Topic}' changed state to '{stateChange.ProducerState}'");
+
     private IProducer<String> CreateProducer(IPulsarClient pulsarClient, string topicName) => pulsarClient.NewProducer(Schema.String)
         .Topic(topicName)
+        .StateChangedHandler(LogState)
         .Create();
 
     private IReader<String> CreateReader(IPulsarClient pulsarClient, MessageId messageId, string topicName) => pulsarClient.NewReader(Schema.String)
         .StartMessageId(messageId)
         .Topic(topicName)
+        .StateChangedHandler(LogState)
         .Create();
 
     private IPulsarClient CreateClient()