Fix Kafka transport check not working bug. #436

src/DotNetCore.CAP.Kafka/IConnectionPool.Default.cs

@@ -24,7 +24,7 @@ namespace DotNetCore.CAP.Kafka
             _producerPool = new ConcurrentQueue<IProducer<string, byte[]>>();
             _maxSize = _options.ConnectionPoolSize;
 
-            logger.LogDebug("Kafka configuration of CAP :\r\n {0}", JsonConvert.SerializeObject(_options.AsKafkaConfig()));
+            logger.LogDebug("CAP Kafka configuration: {0}", JsonConvert.SerializeObject(_options.AsKafkaConfig(), Formatting.Indented));
         }
 
         public string ServersAddress => _options.Servers;

src/DotNetCore.CAP/Internal/IConsumerRegister.Default.cs

@@ -27,8 +27,8 @@ namespace DotNetCore.CAP.Internal
         private readonly TimeSpan _pollingDelay = TimeSpan.FromSeconds(1);
         private readonly CapOptions _options;
         private readonly MethodMatcherCache _selector;
-        private readonly CancellationTokenSource _cts;
 
+        private CancellationTokenSource _cts;
         private string _serverAddress;
         private Task _compositeTask;
         private bool _disposed;
@@ -111,6 +111,7 @@ namespace DotNetCore.CAP.Internal
             {
                 Pulse();
 
+                _cts = new CancellationTokenSource();
                 _isHealthy = true;
 
                 Start();
@@ -247,6 +248,7 @@ namespace DotNetCore.CAP.Internal
                     _logger.LogError("Kafka client consume error. --> " + logmsg.Reason);
                     break;
                 case MqLogType.ServerConnError:
+                    _isHealthy = false;
                     _logger.LogCritical("Kafka server connection error. --> " + logmsg.Reason);
                     break;
                 case MqLogType.ExceptionReceived:

src/DotNetCore.CAP/Processor/IDispatcher.Default.cs

@@ -64,7 +64,11 @@ namespace DotNetCore.CAP.Processor
                         {
                             try
                             {
-                                await _sender.SendAsync(message);
+                                var result = await _sender.SendAsync(message);
+                                if (!result.Succeeded)
+                                {
+                                    _logger.LogWarning(result.Exception, "Message send failed! -->" + result);
+                                }
                             }
                             catch (Exception ex)
                             {

src/DotNetCore.CAP/Processor/IProcessor.Collector.cs

@@ -38,7 +38,7 @@ namespace DotNetCore.CAP.Processor
 
             foreach (var table in tables)
             {
-                _logger.LogDebug($"Collecting expired data from table [{table}].");
+                _logger.LogDebug($"Collecting expired data from table: {table}");
 
                 int deletedCount;
                 var time = DateTime.Now;

src/DotNetCore.CAP/Processor/IProcessor.TransportCheck.cs

@@ -4,16 +4,19 @@
 using System;
 using System.Threading.Tasks;
 using DotNetCore.CAP.Internal;
+using Microsoft.Extensions.Logging;
 
 namespace DotNetCore.CAP.Processor
 {
     public class TransportCheckProcessor : IProcessor
     {
+        private readonly ILogger<TransportCheckProcessor> _logger;
         private readonly IConsumerRegister _register;
         private readonly TimeSpan _waitingInterval;
 
-        public TransportCheckProcessor(IConsumerRegister register)
+        public TransportCheckProcessor(ILogger<TransportCheckProcessor> logger, IConsumerRegister register)
         {
+            _logger = logger;
             _register = register;
             _waitingInterval = TimeSpan.FromSeconds(30);
         }
@@ -25,10 +28,18 @@ namespace DotNetCore.CAP.Processor
                 throw new ArgumentNullException(nameof(context));
             }
 
+            _logger.LogDebug("Transport connection checking...");
+
             if (!_register.IsHealthy())
             {
+                _logger.LogWarning("Transport connection is unhealthy, reconnection...");
+
                 _register.ReStart();
             }
+            else
+            {
+                _logger.LogDebug("Transport connection healthy!");
+            }
 
             await context.WaitAsync(_waitingInterval);
         }