clear out communication client cache from a notification

src/Microsoft.ServiceFabric.Services/Client/IServicePartitionResolver.cs

@@ -19,6 +19,11 @@ namespace Microsoft.ServiceFabric.Services.Client
  /// </summary>
  public interface IServicePartitionResolver
  {
+ /// <summary>
+ /// Event handler that is fired when the service resolver receives a ServiceNotificationFilterMatched Event.
+ /// </summary>
+ event EventHandler<FabricClient.ServiceManagementClient.ServiceNotificationEventArgs> ServiceNotificationEvent;
+
  /// <summary>
  /// Resolves a partition of the specified service with specified back-off/retry settings on retry-able errors.
  /// </summary>

src/Microsoft.ServiceFabric.Services/Client/ServicePartitionResolver.cs

@@ -141,6 +141,11 @@ public ServicePartitionResolver(
  {
  }
 
+ /// <summary>
+ /// Event handler that is fired when the service resolver receives a ServiceNotificationFilterMatched Event.
+ /// </summary>
+ public event EventHandler<FabricClient.ServiceManagementClient.ServiceNotificationEventArgs> ServiceNotificationEvent;
+
  internal bool UseNotification { get; set; }
 
  /// <summary>
@@ -675,12 +680,24 @@ private FabricClient GetClient()
  if (this.fabricClient == null)
  {
  this.fabricClient = this.createFabricClient.Invoke();
+ if (this.UseNotification)
+ {
+ this.fabricClient.ServiceManager.ServiceNotificationFilterMatched += this.OnServiceNotificationFilterMatched;
+ }
  }
 
  return this.fabricClient;
  }
  }
 
+ private void OnServiceNotificationFilterMatched(object sender, EventArgs e)
+ {
+ if (e is FabricClient.ServiceManagementClient.ServiceNotificationEventArgs notificationargs)
+ {
+ this.ServiceNotificationEvent?.Invoke(sender, notificationargs);
+ }
+ }
+
  private void ReportFaulted(FabricClient client)
  {
  lock (this.thisLock)

src/Microsoft.ServiceFabric.Services/Communication/Client/CommunicationClientCache.cs

@@ -10,6 +10,7 @@ namespace Microsoft.ServiceFabric.Services.Communication.Client
  using System.Collections.Generic;
  using System.Fabric;
  using System.Globalization;
+ using System.Linq;
  using System.Threading;
 
  /// <summary>
@@ -93,9 +94,17 @@ public bool TryGetClientCacheEntry(
  return partitionClientCache.TryGetClientCacheEntry(endpoint, listenerName, out cacheEntry);
  }
 
- public void ClearClientCacheEntries(Guid partitionId)
+ /// <summary>
+ /// Clear the cache entries for a given replica that are not contains in validEndpoints
+ /// </summary>
+ /// <param name="partitionId">Partition id of the replica set</param>
+ /// <param name="validEndpoints">Collection of endpoints that are valid for the replica set</param>
+ public void ClearClientCacheEntries(Guid partitionId, ICollection<ResolvedServiceEndpoint> validEndpoints)
  {
- // Currently no op. To implement when we register for service change notifications.
+ if (this.clientCache.TryGetValue(partitionId, out PartitionClientCache clientCache))
+ {
+ clientCache.TryRemoveClientCacheEntry(validEndpoints);
+ }
  }
 
  public void Dispose()
@@ -240,6 +249,26 @@ public bool TryGetClientCacheEntry(
  out cacheEntry);
  }
 
+ /// <summary>
+ /// Try to remove any cache items if the endpoint is not in the collection of valid endpoints
+ /// </summary>
+ /// <param name="validEndpoints">Collection of valid endpoints</param>
+ /// <returns>A boolean indicating if anything was removed from the cache</returns>
+ public bool TryRemoveClientCacheEntry(ICollection<ResolvedServiceEndpoint> validEndpoints)
+ {
+ bool removed = false;
+ foreach (var key in this.cache.Keys)
+ {
+ // if the cache for the replica set contains an endpoint that is not valid remove it
+ if (!validEndpoints.Contains(key.Endpoint))
+ {
+ removed |= this.cache.TryRemove(key, out _);
+ }
+ }
+
+ return removed;
+ }
+
  // Cleaning up of cache entries needs to synchronize with the code that uses the cache entry and sets the
  // communicationClient inside the cache entry. So the removal of the cache-entry should set entry.IsInCache to false
  // and also remove the entry while holding the entry's semaphore, so that we don't purge a cache entry that has a valid

src/Microsoft.ServiceFabric.Services/Communication/Client/CommunicationClientFactoryBase.cs

@@ -41,15 +41,18 @@ public abstract class CommunicationClientFactoryBase<TCommunicationClient> :
  /// <param name="servicePartitionResolver">Optional ServicePartitionResolver</param>
  /// <param name="exceptionHandlers">Optional Custom exception handlers for the exceptions on the Client to Service communication channel</param>
  /// <param name="traceId">Identifier to use in diagnostics traces from this component </param>
+ /// <param name="clearCachedClientsWithNotifications">Indicates that this factory should invalidate cached partition clients through notifications</param>
  protected CommunicationClientFactoryBase(
  IServicePartitionResolver servicePartitionResolver = null,
  IEnumerable<IExceptionHandler> exceptionHandlers = null,
- string traceId = null)
+ string traceId = null,
+ bool clearCachedClientsWithNotifications = false)
  : this(
  false,
  servicePartitionResolver,
  exceptionHandlers,
- traceId)
+ traceId,
+ clearCachedClientsWithNotifications)
  {
  }
 
@@ -60,16 +63,23 @@ protected CommunicationClientFactoryBase(
  /// <param name="servicePartitionResolver">Optional ServicePartitionResolver</param>
  /// <param name="exceptionHandlers">Optional Custom exception handlers for the exceptions on the Client to Service communication channel</param>
  /// <param name="traceId">Identifier to use in diagnostics traces from this component </param>
+ /// <param name="clearCachedClientsWithNotifications">Indicates that this factory should invalidate cached partition clients through notifications</param>
  protected CommunicationClientFactoryBase(
  bool fireConnectEvents,
  IServicePartitionResolver servicePartitionResolver = null,
  IEnumerable<IExceptionHandler> exceptionHandlers = null,
- string traceId = null)
+ string traceId = null,
+ bool clearCachedClientsWithNotifications = false)
  {
  this.fireConnectEvents = fireConnectEvents;
  this.randomLock = new object();
  this.traceId = traceId ?? Guid.NewGuid().ToString();
  this.servicePartitionResolver = servicePartitionResolver ?? ServicePartitionResolver.GetDefault();
+ if (clearCachedClientsWithNotifications)
+ {
+ this.servicePartitionResolver.ServiceNotificationEvent += this.ClearCache;
+ }
+
  this.random = null;
  this.exceptionHandlers = new List<IExceptionHandler>();
  if (exceptionHandlers != null)
@@ -802,6 +812,11 @@ private bool ValidateLockedClientCacheEntry(
  return false;
  }
 
+ private void ClearCache(object sender, FabricClient.ServiceManagementClient.ServiceNotificationEventArgs eventArgs)
+ {
+ this.cache.ClearClientCacheEntries(eventArgs.Notification.PartitionId, eventArgs.Notification.Endpoints);
+ }
+
  private int GenerateSeed()
  {
  var hashcode = this.GetHashCode();