You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@pulsar.apache.org by GitBox <gi...@apache.org> on 2022/05/16 10:57:10 UTC
[GitHub] [pulsar-dotpulsar] goldenccargill opened a new pull request, #104: add server timeout functionality plus docker test
goldenccargill opened a new pull request, #104:
URL: https://github.com/apache/pulsar-dotpulsar/pull/104
As mentioned in the keep alive section of the pulsar binary configuration [here](https://pulsar.apache.org/docs/v2.0.0-rc1-incubating/project/BinaryProtocol/) if the server stops responding to our pings when we have to assume the connection is no longer valid. We can't just rely on the transport to tell us if the connection is alive or not. I've implemented it based on any response from the server rather than specifically a pong, its open to debate if this is correct, I could go either way.
I've also added a test to demonstrate the bug which is fixed by my code in the `PingPongHandler`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-dotpulsar] goldenccargill commented on a diff in pull request #104: add server timeout functionality plus docker test
Posted by GitBox <gi...@apache.org>.
goldenccargill commented on code in PR #104:
URL: https://github.com/apache/pulsar-dotpulsar/pull/104#discussion_r888819170
##########
src/DotPulsar/Abstractions/IPulsarClientBuilder.cs:
##########
@@ -53,6 +53,16 @@ public interface IPulsarClientBuilder
/// </summary>
IPulsarClientBuilder KeepAliveInterval(TimeSpan interval);
+ /// <summary>
+ /// The maximum amount of time to wait without receiving any message from the server at
+ /// which point the connection is assumed to be dead or the server is not responding.
+ /// As we are sending pings the server should respond to those at a minimum within this specified timeout period.
+ /// Once this happens the connection will be torn down and all consumers/producers will enter
+ /// the disconnected state and attempt to reconnect
+ /// The default is 60 seconds.
+ /// </summary>
+ IPulsarClientBuilder ServerResponseTimeout(TimeSpan interval);
Review Comment:
don't know about other clients but it seems it is configurable on the broker so the client setting should agree with that so I think it needs to be configurable here. Shame the value isn't returned from the server as part of the connection response
##########
src/DotPulsar/Internal/PingPongHandler.cs:
##########
@@ -25,29 +25,37 @@ public sealed class PingPongHandler : IAsyncDisposable
{
private readonly IConnection _connection;
private readonly TimeSpan _keepAliveInterval;
+ private readonly TimeSpan _serverResponseTimeout;
private readonly Timer _timer;
private readonly CommandPing _ping;
private readonly CommandPong _pong;
private long _lastCommand;
+ private readonly TaskCompletionSource<object> _serverNotRespondingTcs;
Review Comment:
seems it has been removed
https://github.com/StephenCleary/AsyncEx/issues/176
##########
src/DotPulsar/Internal/PingPongHandler.cs:
##########
@@ -61,13 +69,25 @@ private void Watch(object? state)
var lastCommand = Interlocked.Read(ref _lastCommand);
var now = Stopwatch.GetTimestamp();
var elapsed = TimeSpan.FromSeconds((now - lastCommand) / Stopwatch.Frequency);
- if (elapsed >= _keepAliveInterval)
+
+ if (elapsed > _serverResponseTimeout)
{
- Task.Factory.StartNew(() => SendPing());
- _timer.Change(_keepAliveInterval, TimeSpan.Zero);
+ DotPulsarMeter.ServerTimedout();
+ _serverNotRespondingTcs.SetResult(new object());
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-dotpulsar] goldenccargill commented on a diff in pull request #104: add server timeout functionality plus docker test
Posted by GitBox <gi...@apache.org>.
goldenccargill commented on code in PR #104:
URL: https://github.com/apache/pulsar-dotpulsar/pull/104#discussion_r888819328
##########
tests/DotPulsar.Tests/ProducerTests.cs:
##########
@@ -152,6 +152,7 @@ private IPulsarClient CreateClient()
=> PulsarClient
.Builder()
.Authentication(AuthenticationFactory.Token(ct => ValueTask.FromResult(_fixture.CreateToken(Timeout.InfiniteTimeSpan))))
+ .KeepAliveInterval(TimeSpan.FromSeconds(5))
Review Comment:
to make the test run faster, the check to see if the server responded happens every 5 seconds instead of 30
##########
tests/DotPulsar.Tests/IntegrationCollection.cs:
##########
@@ -18,3 +18,6 @@ namespace DotPulsar.Tests;
[CollectionDefinition("Integration")]
public class IntegrationCollection : ICollectionFixture<IntegrationFixture> { }
+
+[CollectionDefinition("KeepAlive")]
+public class KeepAliveCollection : ICollectionFixture<KeepAliveFixture> { }
Review Comment:
I tried to use the existing cluster but once you start poking around with the network subsequent tests failed so I opted for a separate cluster
##########
tests/DotPulsar.Tests/xunit.runner.json:
##########
@@ -1,4 +1,5 @@
{
"$schema": "https://xunit.net/schema/current/xunit.runner.schema.json",
- "diagnosticMessages": true
+ "diagnosticMessages": true,
+ "parallelizeTestCollections": false
Review Comment:
for same reason as above I can't
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-dotpulsar] blankensteiner commented on a diff in pull request #104: add server timeout functionality plus docker test
Posted by GitBox <gi...@apache.org>.
blankensteiner commented on code in PR #104:
URL: https://github.com/apache/pulsar-dotpulsar/pull/104#discussion_r879466029
##########
tests/DotPulsar.Tests/ProducerTests.cs:
##########
@@ -152,6 +152,7 @@ private IPulsarClient CreateClient()
=> PulsarClient
.Builder()
.Authentication(AuthenticationFactory.Token(ct => ValueTask.FromResult(_fixture.CreateToken(Timeout.InfiniteTimeSpan))))
+ .KeepAliveInterval(TimeSpan.FromSeconds(5))
Review Comment:
Why is this added?
##########
tests/DotPulsar.Tests/xunit.runner.json:
##########
@@ -1,4 +1,5 @@
{
"$schema": "https://xunit.net/schema/current/xunit.runner.schema.json",
- "diagnosticMessages": true
+ "diagnosticMessages": true,
+ "parallelizeTestCollections": false
Review Comment:
This can be removed again if you go ahead with just one cluster for all integration tests.
##########
DotPulsar.sln:
##########
@@ -26,6 +26,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Processing", "samples\Processing\Processing.csproj", "{CC1494FA-4EB5-4DB9-8BE9-0A6E8D0D963E}"
EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DotPulsar.Consumer", "tests\DotPulsar.Consumer\DotPulsar.Consumer.csproj", "{36E6E6EF-A471-4AE4-B696-1C9DAAFA2770}"
Review Comment:
Let's get the new tests into DotPulsar.Tests instead of creating new test projects.
##########
tests/DotPulsar.Tests/IntegrationCollection.cs:
##########
@@ -18,3 +18,6 @@ namespace DotPulsar.Tests;
[CollectionDefinition("Integration")]
public class IntegrationCollection : ICollectionFixture<IntegrationFixture> { }
+
+[CollectionDefinition("KeepAlive")]
+public class KeepAliveCollection : ICollectionFixture<KeepAliveFixture> { }
Review Comment:
Is a new (standalone) cluster needed?
Seems we could solve this with just one cluster/integration fixture.
##########
src/DotPulsar/Abstractions/IPulsarClientBuilder.cs:
##########
@@ -53,6 +53,16 @@ public interface IPulsarClientBuilder
/// </summary>
IPulsarClientBuilder KeepAliveInterval(TimeSpan interval);
+ /// <summary>
+ /// The maximum amount of time to wait without receiving any message from the server at
+ /// which point the connection is assumed to be dead or the server is not responding.
+ /// As we are sending pings the server should respond to those at a minimum within this specified timeout period.
+ /// Once this happens the connection will be torn down and all consumers/producers will enter
+ /// the disconnected state and attempt to reconnect
+ /// The default is 60 seconds.
+ /// </summary>
+ IPulsarClientBuilder ServerResponseTimeout(TimeSpan interval);
Review Comment:
Is this also a configurable setting for other clients? If not, we could just hardcode it.
##########
src/DotPulsar/Internal/PingPongHandler.cs:
##########
@@ -25,29 +25,37 @@ public sealed class PingPongHandler : IAsyncDisposable
{
private readonly IConnection _connection;
private readonly TimeSpan _keepAliveInterval;
+ private readonly TimeSpan _serverResponseTimeout;
private readonly Timer _timer;
private readonly CommandPing _ping;
private readonly CommandPong _pong;
private long _lastCommand;
+ private readonly TaskCompletionSource<object> _serverNotRespondingTcs;
Review Comment:
The non-generic TaskCompletionSource is a better fit since the object is never used.
##########
src/DotPulsar/Internal/PingPongHandler.cs:
##########
@@ -61,13 +69,25 @@ private void Watch(object? state)
var lastCommand = Interlocked.Read(ref _lastCommand);
var now = Stopwatch.GetTimestamp();
var elapsed = TimeSpan.FromSeconds((now - lastCommand) / Stopwatch.Frequency);
- if (elapsed >= _keepAliveInterval)
+
+ if (elapsed > _serverResponseTimeout)
{
- Task.Factory.StartNew(() => SendPing());
- _timer.Change(_keepAliveInterval, TimeSpan.Zero);
+ DotPulsarMeter.ServerTimedout();
+ _serverNotRespondingTcs.SetResult(new object());
Review Comment:
You might as well just return here instead of wrapping the following code in an else.
##########
src/DotPulsar/Internal/Connection.cs:
##########
@@ -294,6 +298,11 @@ private async Task Send(BaseCommand command, CancellationToken cancellationToken
}
public async Task ProcessIncommingFrames(CancellationToken cancellationToken)
+ {
+ await Task.WhenAny(ProcessIncommingFramesImpl(cancellationToken), _pingPongHandler.ServerNotResponding);
Review Comment:
This needs a ConfigureAwait(false)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-dotpulsar] goldenccargill commented on a diff in pull request #104: add server timeout functionality plus docker test
Posted by GitBox <gi...@apache.org>.
goldenccargill commented on code in PR #104:
URL: https://github.com/apache/pulsar-dotpulsar/pull/104#discussion_r888819496
##########
src/DotPulsar/Internal/Connection.cs:
##########
@@ -294,6 +298,11 @@ private async Task Send(BaseCommand command, CancellationToken cancellationToken
}
public async Task ProcessIncommingFrames(CancellationToken cancellationToken)
+ {
+ await Task.WhenAny(ProcessIncommingFramesImpl(cancellationToken), _pingPongHandler.ServerNotResponding);
Review Comment:
done
##########
DotPulsar.sln:
##########
@@ -26,6 +26,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Processing", "samples\Processing\Processing.csproj", "{CC1494FA-4EB5-4DB9-8BE9-0A6E8D0D963E}"
EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DotPulsar.Consumer", "tests\DotPulsar.Consumer\DotPulsar.Consumer.csproj", "{36E6E6EF-A471-4AE4-B696-1C9DAAFA2770}"
Review Comment:
I need an executable that I can run in docker to connect to the server
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org