You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by mc...@apache.org on 2019/04/22 22:20:05 UTC
[geode] branch develop updated: GEODE-6607: Moving client
registration queue to CacheClientNotifier
This is an automated email from the ASF dual-hosted git repository.
mcmellawatt pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/geode.git
The following commit(s) were added to refs/heads/develop by this push:
new afc311c GEODE-6607: Moving client registration queue to CacheClientNotifier
afc311c is described below
commit afc311c04f6908a8f725834cdf2c49ce6e902b3f
Author: Ryan McMahon <rm...@pivotal.io>
AuthorDate: Thu Apr 18 10:30:29 2019 -0700
GEODE-6607: Moving client registration queue to CacheClientNotifier
To avoid client subscription data inconsistencies, we need to ensure
that we minimize the chance that an event is processed while a client is
registering but before it has fully registered. There are two major
phases in registration - one is to request filter info from a peer
already hosting the queue for the client, and the other is doing a GII
of the queue from a peer. If an event which a client would be
interested in is processed concurrently during registration, but before
the filter info has been fully received and processed, the event will be
missed by the client. To reduce this window, we will start queueing
events for the registering client as soon as possible (deserialization
of the client proxy membership ID). After registration is complete, we
drain the queued events and put them into the clients subscription
queue.
To make this code unit testable, it was necessary to extract the logic
reading data off the socket/deserializing that data into a separate
class which can be injected, the ClientRegistrationMetadata class. This
allows us to mock a ClientRegistrationMetadata without actually doing any IO.
The CacheClientNotifier could be futher broken up to allow for even more
unit testability, but this was a first step in the right direction.
Co-authored-by: Ryan McMahon <rm...@pivotal.io>
Co-authored-by: Murtuza Boxwala <mb...@pivotal.io>
Co-authored-by: Ernie Burghardt <eb...@pivotal.io>
---
.../CacheClientNotifierIntegrationTest.java | 301 ----------
.../internal/cache/tier/sockets/AcceptorImpl.java | 12 +-
.../cache/tier/sockets/CacheClientNotifier.java | 619 +++++++++------------
.../cache/tier/sockets/CacheClientProxy.java | 88 +--
.../ClientRegistrationEventQueueManager.java | 279 ++++++++++
.../tier/sockets/ClientRegistrationMetadata.java | 193 +++++++
.../cache/tier/sockets/SocketMessageWriter.java | 102 ++++
.../tier/sockets/CacheClientNotifierTest.java | 176 ++++++
.../ClientRegistrationEventQueueManagerTest.java | 199 +++++++
.../java/org/apache/geode/test/fake/Fakes.java | 9 +-
10 files changed, 1239 insertions(+), 739 deletions(-)
diff --git a/geode-core/src/integrationTest/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifierIntegrationTest.java b/geode-core/src/integrationTest/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifierIntegrationTest.java
deleted file mode 100644
index 774ba05..0000000
--- a/geode-core/src/integrationTest/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifierIntegrationTest.java
+++ /dev/null
@@ -1,301 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
- * agreements. See the NOTICE file distributed with this work for additional information regarding
- * copyright ownership. The ASF licenses this file to You 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
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software distributed under the License
- * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
- * or implied. See the License for the specific language governing permissions and limitations under
- * the License.
- */
-package org.apache.geode.internal.cache.tier.sockets;
-
-import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
-import static org.apache.geode.test.awaitility.GeodeAwaitility.getTimeout;
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyString;
-import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.Mockito.doAnswer;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.spy;
-import static org.mockito.Mockito.when;
-
-import java.io.IOException;
-import java.net.InetAddress;
-import java.net.Socket;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.Callable;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-
-import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
-import org.apache.shiro.subject.Subject;
-import org.junit.After;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-import org.junit.rules.Timeout;
-
-import org.apache.geode.CancelCriterion;
-import org.apache.geode.Statistics;
-import org.apache.geode.StatisticsType;
-import org.apache.geode.cache.AttributesMutator;
-import org.apache.geode.cache.Operation;
-import org.apache.geode.cache.RegionAttributes;
-import org.apache.geode.distributed.DistributedMember;
-import org.apache.geode.distributed.internal.DSClock;
-import org.apache.geode.distributed.internal.DistributionConfig;
-import org.apache.geode.distributed.internal.DistributionManager;
-import org.apache.geode.distributed.internal.InternalDistributedSystem;
-import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
-import org.apache.geode.internal.SystemTimer;
-import org.apache.geode.internal.Version;
-import org.apache.geode.internal.cache.EntryEventImpl;
-import org.apache.geode.internal.cache.EnumListenerEvent;
-import org.apache.geode.internal.cache.EventID;
-import org.apache.geode.internal.cache.FilterProfile;
-import org.apache.geode.internal.cache.FilterRoutingInfo;
-import org.apache.geode.internal.cache.HARegion;
-import org.apache.geode.internal.cache.InternalCache;
-import org.apache.geode.internal.cache.InternalRegionArguments;
-import org.apache.geode.internal.cache.LocalRegion;
-import org.apache.geode.internal.security.SecurityService;
-import org.apache.geode.test.junit.categories.ClientSubscriptionTest;
-
-@Category(ClientSubscriptionTest.class)
-public class CacheClientNotifierIntegrationTest {
-
- private final CountDownLatch messageDispatcherInitLatch = new CountDownLatch(1);
- private final CountDownLatch notifyClientLatch = new CountDownLatch(1);
-
- private final ExecutorService executorService = Executors.newFixedThreadPool(2);
-
- @Rule
- public Timeout timeout = Timeout.millis(getTimeout().getValueInMS());
-
- @After
- public void tearDown() {
- messageDispatcherInitLatch.countDown();
- notifyClientLatch.countDown();
- executorService.shutdownNow();
- }
-
- @Test
- public void testCacheClientNotifier_NotifyClients_QRMCausesPrematureRemovalFromHAContainer()
- throws Exception {
- InternalCache mockInternalCache = createMockInternalCache();
- assertThat(mockInternalCache.getMeterRegistry()).isNotNull();
-
- CacheClientNotifier cacheClientNotifier =
- CacheClientNotifier.getInstance(mockInternalCache, mock(CacheServerStats.class),
- 100000, 100000, mock(ConnectionListener.class), null, false);
-
- String mockRegionNameProxyOne = "mockHARegionProxyOne";
- String mockRegionNameProxyTwo = "mockHARegionProxyTwo";
-
- CacheClientProxy cacheClientProxyOne =
- createMockCacheClientProxy(cacheClientNotifier, mockRegionNameProxyOne);
-
- CacheClientProxy cacheClientProxyTwo =
- createMockCacheClientProxy(cacheClientNotifier, mockRegionNameProxyTwo);
-
- doAnswer(invocation -> {
- messageDispatcherInitLatch.countDown();
- notifyClientLatch.await();
- return invocation.callRealMethod();
- }).when(cacheClientProxyTwo).createMessageDispatcher(anyString());
-
- createMockHARegion(mockInternalCache, cacheClientProxyOne, mockRegionNameProxyOne, true);
- createMockHARegion(mockInternalCache, cacheClientProxyTwo, mockRegionNameProxyTwo, false);
-
- List<Callable<Void>> initAndNotifyTasks = new ArrayList<>();
-
- // On one thread, we are initializing the message dispatchers for the two CacheClientProxy
- // objects. For the second client's initialization,
- // we block until we can simulate a QRM message while the event sits in the CacheClientProxy's
- // queuedEvents collection.
- // This blocking logic can be found in setupMockMessageDispatcher().
- initAndNotifyTasks.add(() -> {
- cacheClientProxyOne.initializeMessageDispatcher();
- cacheClientProxyTwo.initializeMessageDispatcher();
- return null;
- });
-
- EntryEventImpl mockEntryEventImpl = createMockEntryEvent(new ArrayList<CacheClientProxy>() {
- {
- add(cacheClientProxyOne);
- add(cacheClientProxyTwo);
- }
- });
-
- // On a second thread, we are notifying clients of our mocked events. For the first client, we
- // wait for the message dispatcher to be
- // initialized so the event is put/processed. For the second client, we ensure that the
- // dispatcher is still initializing so we put the event
- // in the CacheClientProxy's queuedEvents collection. When the event is handled by the first
- // CacheClientProxy, we mocked a QRM in the HARegion
- // put to simulate a QRM message being received at that time. The simulated QRM logic can be
- // found in the createMockHARegion() method.
- initAndNotifyTasks.add(() -> {
- messageDispatcherInitLatch.await();
- CacheClientNotifier.notifyClients(mockEntryEventImpl);
- notifyClientLatch.countDown();
- return null;
- });
-
- List<Future<Void>> futures = executorService.invokeAll(initAndNotifyTasks);
-
- for (Future<Void> future : futures) {
- future.get();
- }
-
- // Verify that we do not hang in peek() for the second proxy due to the wrapper
- await().until(() -> {
- if (cacheClientProxyTwo.getHARegionQueue().peek() != null) {
- cacheClientProxyTwo.getHARegionQueue().remove();
- return true;
- }
- return false;
- });
-
- assertThat(cacheClientNotifier.getHaContainer()).as("Expected the HAContainer to be empty")
- .isEmpty();
- }
-
- private void createMockHARegion(InternalCache internalCache,
- CacheClientProxy cacheClientProxy, String haRegionName, boolean simulateQrm)
- throws IOException, ClassNotFoundException {
- HARegion mockHARegion = mock(HARegion.class);
-
- when(mockHARegion.getAttributesMutator()).thenReturn(mock(AttributesMutator.class));
- when(mockHARegion.getCache()).thenReturn(internalCache);
- when(mockHARegion.getCancelCriterion()).thenReturn(mock(CancelCriterion.class));
- when(mockHARegion.getGemFireCache()).thenReturn(internalCache);
- when(internalCache.createVMRegion(eq(haRegionName), any(RegionAttributes.class),
- any(InternalRegionArguments.class))).thenReturn(mockHARegion);
-
- // We use a mock of the HARegion.put() method to simulate an queue removal message
- // immediately after the event was successfully put. In production when a queue removal takes
- // place at this time,
- // it will decrement the ref count on the HAEventWrapper and potentially make it eligible for
- // removal later on
- // when CacheClientNotifier.checkAndRemoveFromClientMsgsRegion() is called.
-
- Map<Object, Object> events = new HashMap<>();
-
- when(mockHARegion.put(any(long.class), any(HAEventWrapper.class)))
- .thenAnswer(invocation -> {
- long position = invocation.getArgument(0);
- HAEventWrapper haEventWrapper = invocation.getArgument(1);
-
- if (simulateQrm) {
- // This call is ultimately what a QRM message will do when it is processed, so we
- // simulate
- // that here.
- cacheClientProxy.getHARegionQueue().destroyFromAvailableIDs(position);
- events.remove(position);
- cacheClientProxy.getHARegionQueue()
- .decAndRemoveFromHAContainer(haEventWrapper);
- } else {
- events.put(position, haEventWrapper);
- }
-
- return null;
- });
-
- // This is so that when peek() is called, the object is returned. Later we want to verify that
- // it was successfully "delivered" to the client and subsequently removed from the HARegion.
- when(mockHARegion.get(any(long.class)))
- .thenAnswer(invocation -> events.get(invocation.getArgument(0)));
- }
-
- private InternalCache createMockInternalCache() {
- InternalCache mockInternalCache = mock(InternalCache.class);
- InternalDistributedSystem mockInternalDistributedSystem = createMockInternalDistributedSystem();
-
- when(mockInternalCache.getCancelCriterion()).thenReturn(mock(CancelCriterion.class));
- when(mockInternalCache.getCCPTimer()).thenReturn(mock(SystemTimer.class));
- when(mockInternalCache.getDistributedSystem()).thenReturn(mockInternalDistributedSystem);
- when(mockInternalCache.getInternalDistributedSystem())
- .thenReturn(mockInternalDistributedSystem);
- when(mockInternalCache.getMeterRegistry()).thenReturn(new SimpleMeterRegistry());
-
- return mockInternalCache;
- }
-
- private InternalDistributedSystem createMockInternalDistributedSystem() {
- DistributionManager mockDistributionManager = mock(DistributionManager.class);
- InternalDistributedSystem mockInternalDistributedSystem = mock(InternalDistributedSystem.class);
-
- when(mockDistributionManager.getConfig()).thenReturn(mock(DistributionConfig.class));
- when(mockInternalDistributedSystem.getDistributedMember())
- .thenReturn(mock(InternalDistributedMember.class));
- when(mockInternalDistributedSystem.createAtomicStatistics(any(StatisticsType.class),
- any(String.class))).thenReturn(mock(Statistics.class));
- when(mockInternalDistributedSystem.getDistributionManager())
- .thenReturn(mockDistributionManager);
- when(mockInternalDistributedSystem.getClock()).thenReturn(mock(DSClock.class));
-
- return mockInternalDistributedSystem;
- }
-
- private CacheClientProxy createMockCacheClientProxy(CacheClientNotifier cacheClientNotifier,
- String haRegionName) throws IOException {
- ClientProxyMembershipID mockClientProxyMembershipID = mock(ClientProxyMembershipID.class);
- Socket mockSocket = mock(Socket.class);
-
- when(mockClientProxyMembershipID.getDistributedMember())
- .thenReturn(mock(DistributedMember.class));
- when(mockClientProxyMembershipID.getHARegionName()).thenReturn(haRegionName);
- when(mockSocket.getInetAddress()).thenReturn(mock(InetAddress.class));
-
- CacheClientProxy cacheClientProxy =
- spy(new CacheClientProxy(cacheClientNotifier, mockSocket, mockClientProxyMembershipID, true,
- (byte) 0, Version.GFE_58, 0, true, mock(SecurityService.class), mock(Subject.class)));
-
- cacheClientNotifier.addClientInitProxy(cacheClientProxy);
-
- return cacheClientProxy;
- }
-
- private EntryEventImpl createMockEntryEvent(List<CacheClientProxy> proxies) {
- EntryEventImpl mockEntryEventImpl = mock(EntryEventImpl.class);
- EventID mockEventId = mock(EventID.class);
- FilterProfile mockFilterProfile = mock(FilterProfile.class);
- FilterRoutingInfo.FilterInfo mockFilterInfo = mock(FilterRoutingInfo.FilterInfo.class);
- LocalRegion mockLocalRegion = mock(LocalRegion.class);
- Set mockInterestedClients = mock(Set.class);
-
- when(mockEntryEventImpl.getEventId()).thenReturn(mockEventId);
- when(mockEntryEventImpl.getEventType()).thenReturn(EnumListenerEvent.AFTER_CREATE);
- when(mockEntryEventImpl.getLocalFilterInfo()).thenReturn(mockFilterInfo);
- when(mockEntryEventImpl.getOperation()).thenReturn(Operation.CREATE);
- when(mockEntryEventImpl.getRegion()).thenReturn(mockLocalRegion);
- when(mockEventId.getMembershipID()).thenReturn(new byte[] {1});
- when(mockFilterInfo.getInterestedClients()).thenReturn(mockInterestedClients);
- when(mockFilterInfo.getInterestedClientsInv()).thenReturn(null);
-
- Set<ClientProxyMembershipID> mockRealClientIDs = new HashSet<>();
- for (CacheClientProxy proxy : proxies) {
- mockRealClientIDs.add(proxy.getProxyID());
- }
-
- when(mockFilterProfile.getRealClientIDs(any(Collection.class))).thenReturn(mockRealClientIDs);
- when(mockLocalRegion.getFilterProfile()).thenReturn(mockFilterProfile);
-
- return mockEntryEventImpl;
- }
-}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImpl.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImpl.java
index 7cfc7d2..87ed673 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImpl.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/AcceptorImpl.java
@@ -1840,9 +1840,15 @@ public class AcceptorImpl implements Acceptor, Runnable, CommBufferPool {
logger.info(":Cache server: Initializing {} server-to-client communication socket: {}",
isPrimaryServerToClient ? "primary" : "secondary", socket);
try {
- acceptor.getCacheClientNotifier().registerClient(socket, isPrimaryServerToClient,
- acceptor.getAcceptorId(), acceptor.isNotifyBySubscription());
- } catch (IOException ex) {
+ ClientRegistrationMetadata clientRegistrationMetadata =
+ new ClientRegistrationMetadata(acceptor.cache, socket);
+
+ if (clientRegistrationMetadata.initialize()) {
+ acceptor.getCacheClientNotifier().registerClient(clientRegistrationMetadata, socket,
+ isPrimaryServerToClient,
+ acceptor.getAcceptorId(), acceptor.isNotifyBySubscription());
+ }
+ } catch (final IOException ex) {
closeSocket(socket);
if (acceptor.isRunning()) {
if (!acceptor.loggedAcceptError) {
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifier.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifier.java
index f96bad7..a7dc8a4 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifier.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifier.java
@@ -18,12 +18,11 @@ import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_CLIE
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_CLIENT_AUTHENTICATOR;
import java.io.BufferedOutputStream;
-import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.Socket;
-import java.net.SocketAddress;
import java.security.Principal;
import java.util.ArrayList;
import java.util.Collection;
@@ -36,17 +35,17 @@ import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.logging.log4j.Logger;
import org.apache.shiro.subject.Subject;
import org.apache.geode.CancelException;
-import org.apache.geode.DataSerializer;
-import org.apache.geode.Instantiator;
import org.apache.geode.InternalGemFireError;
import org.apache.geode.StatisticsFactory;
import org.apache.geode.annotations.internal.MakeNotStatic;
@@ -57,7 +56,6 @@ import org.apache.geode.cache.InterestRegistrationListener;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.RegionExistsException;
-import org.apache.geode.cache.UnsupportedVersionException;
import org.apache.geode.cache.client.internal.PoolImpl;
import org.apache.geode.cache.client.internal.PoolImpl.PoolTask;
import org.apache.geode.cache.query.CqException;
@@ -72,12 +70,8 @@ import org.apache.geode.distributed.internal.DistributionConfig;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.internal.ClassLoadUtil;
-import org.apache.geode.internal.InternalDataSerializer;
-import org.apache.geode.internal.InternalInstantiator;
import org.apache.geode.internal.SystemTimer;
import org.apache.geode.internal.Version;
-import org.apache.geode.internal.VersionedDataInputStream;
-import org.apache.geode.internal.VersionedDataOutputStream;
import org.apache.geode.internal.cache.CacheClientStatus;
import org.apache.geode.internal.cache.CacheDistributionAdvisor;
import org.apache.geode.internal.cache.CacheServerImpl;
@@ -111,6 +105,7 @@ import org.apache.geode.internal.statistics.DummyStatisticsFactory;
import org.apache.geode.security.AccessControl;
import org.apache.geode.security.AuthenticationFailedException;
import org.apache.geode.security.AuthenticationRequiredException;
+import org.apache.geode.security.GemFireSecurityException;
/**
* Class <code>CacheClientNotifier</code> works on the server and manages client socket connections
@@ -121,6 +116,9 @@ import org.apache.geode.security.AuthenticationRequiredException;
@SuppressWarnings({"synthetic-access", "deprecation"})
public class CacheClientNotifier {
private static final Logger logger = LogService.getLogger();
+ private final SocketMessageWriter socketMessageWriter = new SocketMessageWriter();
+ private final ClientRegistrationEventQueueManager registrationQueueManager =
+ new ClientRegistrationEventQueueManager();
@MakeNotStatic
private static volatile CacheClientNotifier ccnSingleton;
@@ -136,7 +134,7 @@ public class CacheClientNotifier {
ConnectionListener listener, List overflowAttributesList, boolean isGatewayReceiver) {
if (ccnSingleton == null) {
ccnSingleton = new CacheClientNotifier(cache, acceptorStats, maximumMessageCount,
- messageTimeToLive, listener, overflowAttributesList, isGatewayReceiver);
+ messageTimeToLive, listener, isGatewayReceiver);
}
if (!isGatewayReceiver && ccnSingleton.getHaContainer() == null) {
@@ -152,269 +150,101 @@ public class CacheClientNotifier {
}
/**
- * Writes a given message to the output stream
- *
- * @param dos the <code>DataOutputStream</code> to use for writing the message
- * @param type a byte representing the message type
- * @param p_msg the message to be written; can be null
- */
- private void writeMessage(DataOutputStream dos, byte type, String p_msg, Version clientVersion)
- throws IOException {
- writeHandshakeMessage(dos, type, p_msg, clientVersion, (byte) 0x00, 0);
- }
-
- private void writeHandshakeMessage(DataOutputStream dos, byte type, String p_msg,
- Version clientVersion, byte endpointType, int queueSize) throws IOException {
- String msg = p_msg;
-
- // write the message type
- dos.writeByte(type);
-
- dos.writeByte(endpointType);
- dos.writeInt(queueSize);
-
- if (msg == null) {
- msg = "";
- }
- dos.writeUTF(msg);
- if (clientVersion != null && clientVersion.compareTo(Version.GFE_61) >= 0) {
- // get all the instantiators.
- Instantiator[] instantiators = InternalInstantiator.getInstantiators();
- HashMap instantiatorMap = new HashMap();
- if (instantiators != null && instantiators.length > 0) {
- for (Instantiator instantiator : instantiators) {
- ArrayList instantiatorAttributes = new ArrayList();
- instantiatorAttributes.add(instantiator.getClass().toString().substring(6));
- instantiatorAttributes.add(instantiator.getInstantiatedClass().toString().substring(6));
- instantiatorMap.put(instantiator.getId(), instantiatorAttributes);
- }
- }
- DataSerializer.writeHashMap(instantiatorMap, dos);
-
- // get all the dataserializers.
- DataSerializer[] dataSerializers = InternalDataSerializer.getSerializers();
- HashMap<Integer, ArrayList<String>> dsToSupportedClasses =
- new HashMap<Integer, ArrayList<String>>();
- HashMap<Integer, String> dataSerializersMap = new HashMap<Integer, String>();
- if (dataSerializers != null && dataSerializers.length > 0) {
- for (DataSerializer dataSerializer : dataSerializers) {
- dataSerializersMap.put(dataSerializer.getId(),
- dataSerializer.getClass().toString().substring(6));
- if (clientVersion.compareTo(Version.GFE_6516) >= 0) {
- ArrayList<String> supportedClassNames = new ArrayList<String>();
- for (Class clazz : dataSerializer.getSupportedClasses()) {
- supportedClassNames.add(clazz.getName());
- }
- dsToSupportedClasses.put(dataSerializer.getId(), supportedClassNames);
- }
- }
- }
- DataSerializer.writeHashMap(dataSerializersMap, dos);
- if (clientVersion.compareTo(Version.GFE_6516) >= 0) {
- DataSerializer.writeHashMap(dsToSupportedClasses, dos);
- }
- if (clientVersion.compareTo(Version.GEODE_1_5_0) >= 0) {
- dos.writeInt(CLIENT_PING_TASK_PERIOD);
- }
- }
- dos.flush();
- }
-
- /**
- * Writes an exception message to the socket
- *
- * @param dos the <code>DataOutputStream</code> to use for writing the message
- * @param type a byte representing the exception type
- * @param ex the exception to be written; should not be null
- */
- private void writeException(DataOutputStream dos, byte type, Exception ex, Version clientVersion)
- throws IOException {
- writeMessage(dos, type, ex.toString(), clientVersion);
- }
-
- /**
* Registers a new client updater that wants to receive updates with this server.
*
* @param socket The socket over which the server communicates with the client.
- */
- public void registerClient(Socket socket, boolean isPrimary, long acceptorId,
- boolean notifyBySubscription) throws IOException {
- // Since no remote ports were specified in the message, wait for them.
+ * @param isPrimary Whether server is the primary subscription end point for this client
+ * @param acceptorId ID of the acceptor used to clean up the client connection
+ * @param notifyBySubscription Whether the Server is running in NotifyBySubscription mode
+ * @throws IOException Can occur if there are issues communicating over the socket
+ */
+ public void registerClient(final ClientRegistrationMetadata clientRegistrationMetadata,
+ final Socket socket, final boolean isPrimary, final long acceptorId,
+ final boolean notifyBySubscription) throws IOException {
long startTime = this.statistics.startTime();
- DataInputStream dis = new DataInputStream(socket.getInputStream());
- DataOutputStream dos = new DataOutputStream(socket.getOutputStream());
-
- // Read the client version
- short clientVersionOrdinal = Version.readOrdinal(dis);
- Version clientVersion = null;
- try {
- clientVersion = Version.fromOrdinal(clientVersionOrdinal, true);
- if (logger.isDebugEnabled()) {
- logger.debug("{}: Registering client with version: {}", this, clientVersion);
- }
- } catch (UnsupportedVersionException e) {
- SocketAddress sa = socket.getRemoteSocketAddress();
- UnsupportedVersionException uve = e;
- if (sa != null) {
- String sInfo = " Client: " + sa.toString() + ".";
- uve = new UnsupportedVersionException(e.getMessage() + sInfo);
- }
- logger.warn("CacheClientNotifier: Caught exception attempting to client: ",
- uve);
- writeException(dos, CommunicationMode.UnsuccessfulServerToClient.getModeNumber(), uve,
- clientVersion);
- return;
- }
- // Read and ignore the reply code. This is used on the client to server
- // handshake.
- dis.readByte(); // replyCode
+ ClientProxyMembershipID clientProxyMembershipID =
+ clientRegistrationMetadata.getClientProxyMembershipID();
+ DataOutputStream dataOutputStream = clientRegistrationMetadata.getDataOutputStream();
+ Version clientVersion = clientRegistrationMetadata.getClientVersion();
- if (Version.GFE_57.compareTo(clientVersion) <= 0) {
- if (Version.CURRENT.compareTo(clientVersion) > 0) {
- dis = new VersionedDataInputStream(dis, clientVersion);
- dos = new VersionedDataOutputStream(dos, clientVersion);
- }
- registerGFEClient(dis, dos, socket, isPrimary, startTime, clientVersion, acceptorId,
- notifyBySubscription);
- } else {
- Exception e = new UnsupportedVersionException(clientVersionOrdinal);
- throw new IOException(e.toString());
- }
- }
-
- protected void registerGFEClient(DataInputStream dis, DataOutputStream dos, Socket socket,
- boolean isPrimary, long startTime, Version clientVersion, long acceptorId,
- boolean notifyBySubscription) throws IOException {
- // Read the ports and throw them away. We no longer need them
- int numberOfPorts = dis.readInt();
- for (int i = 0; i < numberOfPorts; i++) {
- dis.readInt();
- }
- // Read the handshake identifier and convert it to a string member id
- ClientProxyMembershipID proxyID = null;
- CacheClientProxy proxy;
- AccessControl authzCallback = null;
- byte clientConflation;
try {
- proxyID = ClientProxyMembershipID.readCanonicalized(dis);
- if (getDenylistedClient().contains(proxyID)) {
- writeException(dos, Handshake.REPLY_INVALID,
- new Exception("This client is denylisted by server"), clientVersion);
- return;
- }
- proxy = getClientProxy(proxyID);
- DistributedMember member = proxyID.getDistributedMember();
-
- DistributedSystem system = this.getCache().getDistributedSystem();
- Properties sysProps = system.getProperties();
- String authenticator = sysProps.getProperty(SECURITY_CLIENT_AUTHENTICATOR);
-
- if (clientVersion.compareTo(Version.GFE_603) >= 0) {
- byte[] overrides = Handshake.extractOverrides(new byte[] {(byte) dis.read()});
- clientConflation = overrides[0];
- } else {
- clientConflation = (byte) dis.read();
- }
-
- switch (clientConflation) {
- case Handshake.CONFLATION_DEFAULT:
- case Handshake.CONFLATION_OFF:
- case Handshake.CONFLATION_ON:
- break;
- default:
- writeException(dos, Handshake.REPLY_INVALID,
- new IllegalArgumentException("Invalid conflation byte"), clientVersion);
- return;
- }
- Object subject = null;
- Properties credentials =
- Handshake.readCredentials(dis, dos, system, this.cache.getSecurityService());
- if (credentials != null) {
- if (securityLogWriter.fineEnabled()) {
- securityLogWriter
- .fine("CacheClientNotifier: verifying credentials for proxyID: " + proxyID);
- }
- subject =
- Handshake.verifyCredentials(authenticator, credentials, system.getSecurityProperties(),
- this.logWriter, this.securityLogWriter, member, this.cache.getSecurityService());
- }
-
- Subject shiroSubject =
- subject instanceof Subject ? (Subject) subject : null;
- proxy = registerClient(socket, proxyID, proxy, isPrimary, clientConflation, clientVersion,
- acceptorId, notifyBySubscription, shiroSubject);
-
- if (proxy != null && subject != null) {
- if (subject instanceof Principal) {
- Principal principal = (Principal) subject;
- if (securityLogWriter.fineEnabled()) {
- securityLogWriter
- .fine("CacheClientNotifier: successfully verified credentials for proxyID: "
- + proxyID + " having principal: " + principal.getName());
- }
-
- String postAuthzFactoryName = sysProps.getProperty(SECURITY_CLIENT_ACCESSOR_PP);
- if (postAuthzFactoryName != null && postAuthzFactoryName.length() > 0) {
- Method authzMethod = ClassLoadUtil.methodFromName(postAuthzFactoryName);
- authzCallback = (AccessControl) authzMethod.invoke(null, (Object[]) null);
- authzCallback.init(principal, member, this.getCache());
- }
- proxy.setPostAuthzCallback(authzCallback);
+ if (isClientPermitted(clientRegistrationMetadata, clientProxyMembershipID)) {
+ registrationQueueManager.create(clientProxyMembershipID, new ConcurrentLinkedQueue<>(),
+ new ReentrantReadWriteLock());
+
+ try {
+ registerClientInternal(clientRegistrationMetadata, socket, isPrimary, acceptorId,
+ notifyBySubscription);
+ } finally {
+ registrationQueueManager.drain(
+ clientProxyMembershipID,
+ this);
}
}
- } catch (ClassNotFoundException e) {
- throw new IOException(
- String.format(
- "ClientProxyMembershipID object could not be created. Exception occurred was %s",
- e));
- } catch (AuthenticationRequiredException ex) {
- securityLogWriter.warning(
- String.format("An exception was thrown for client [%s]. %s",
- new Object[] {proxyID, ex}));
- writeException(dos, Handshake.REPLY_EXCEPTION_AUTHENTICATION_REQUIRED, ex, clientVersion);
- return;
- } catch (AuthenticationFailedException ex) {
- securityLogWriter.warning(
- String.format("An exception was thrown for client [%s]. %s",
- new Object[] {proxyID, ex}));
- writeException(dos, Handshake.REPLY_EXCEPTION_AUTHENTICATION_FAILED, ex, clientVersion);
- return;
- } catch (CacheException e) {
+ } catch (final AuthenticationRequiredException ex) {
+ handleAuthenticationException(clientProxyMembershipID, dataOutputStream, clientVersion,
+ ex, Handshake.REPLY_EXCEPTION_AUTHENTICATION_REQUIRED);
+ } catch (final AuthenticationFailedException ex) {
+ handleAuthenticationException(clientProxyMembershipID, dataOutputStream, clientVersion, ex,
+ Handshake.REPLY_EXCEPTION_AUTHENTICATION_FAILED);
+ } catch (final CacheException e) {
logger.warn(String.format("%s :registerClient: Exception encountered in registration %s",
- new Object[] {this, e}),
+ this, e),
e);
- IOException io = new IOException(
+ IOException ioException = new IOException(
String.format("Exception occurred while trying to register interest due to : %s",
- e.getMessage()));
- io.initCause(e);
- throw io;
- } catch (Exception ex) {
- logger.warn(String.format("An exception was thrown for client [%s]. %s",
- new Object[] {proxyID, ""}),
+ e.getMessage()),
+ e);
+ throw ioException;
+ } catch (final Exception ex) {
+ logger.warn(String.format("An exception was thrown for client [%s].",
+ (clientProxyMembershipID != null
+ ? clientProxyMembershipID : "unknown")),
ex);
- writeException(dos, CommunicationMode.UnsuccessfulServerToClient.getModeNumber(), ex,
+ socketMessageWriter.writeException(dataOutputStream,
+ CommunicationMode.UnsuccessfulServerToClient.getModeNumber(),
+ ex,
clientVersion);
- return;
}
this.statistics.endClientRegistration(startTime);
}
/**
- * Registers a new client that wants to receive updates with this server.
+ * Continues the registration of a new client that wants to receive updates with this server.
*
+ * @param clientRegistrationMetadata Contains registration info pertaining to the client
* @param socket The socket over which the server communicates with the client.
- * @param proxyId The distributed member id of the client being registered
- * @param proxy The <code>CacheClientProxy</code> of the given <code>proxyId</code>
- *
- * @return CacheClientProxy for the registered client
- */
- private CacheClientProxy registerClient(Socket socket, ClientProxyMembershipID proxyId,
- CacheClientProxy proxy, boolean isPrimary, byte clientConflation, Version clientVersion,
- long acceptorId, boolean notifyBySubscription, Subject subject)
- throws IOException, CacheException {
- CacheClientProxy l_proxy = proxy;
+ * @param isPrimary Whether server is the primary subscription end point for this client
+ * @param acceptorId ID of the acceptor used to clean up the client connection
+ * @param notifyBySubscription Whether the Server is running in NotifyBySubscription mode
+ * @throws IOException Can occur if there are issues communicating over the socket
+ * @throws CacheException A generic exception, which indicates a cache error has occurred.
+ * @throws ClassNotFoundException Thrown when the ClientProxyMembershipID class is not found
+ * @throws NoSuchMethodException Potentially thrown by performPostAuthorization
+ * @throws InvocationTargetException Potentially thrown by performPostAuthorization
+ * @throws IllegalAccessException Potentially thrown by performPostAuthorization
+ */
+ void registerClientInternal(final ClientRegistrationMetadata clientRegistrationMetadata,
+ final Socket socket,
+ final boolean isPrimary,
+ final long acceptorId, final boolean notifyBySubscription)
+ throws IOException, CacheException, ClassNotFoundException, NoSuchMethodException,
+ InvocationTargetException, IllegalAccessException {
+ ClientProxyMembershipID clientProxyMembershipID =
+ clientRegistrationMetadata.getClientProxyMembershipID();
+ byte clientConflation = clientRegistrationMetadata.getClientConflation();
+ Version clientVersion = clientRegistrationMetadata.getClientVersion();
+
+ CacheClientProxy cacheClientProxy = getClientProxy(clientProxyMembershipID);
+ DistributedMember member = clientProxyMembershipID.getDistributedMember();
+ DistributedSystem system = getCache().getDistributedSystem();
+ Properties sysProps = system.getProperties();
+ String authenticator = sysProps.getProperty(SECURITY_CLIENT_AUTHENTICATOR);
+ Object subjectOrPrincipal =
+ getSubjectOrPrincipal(clientRegistrationMetadata, member, system, authenticator);
+ Subject subject = subjectOrPrincipal instanceof Subject ? (Subject) subjectOrPrincipal : null;
// Initialize the socket
socket.setTcpNoDelay(true);
@@ -431,11 +261,11 @@ public class CacheClientNotifier {
byte responseByte = CommunicationMode.SuccessfulServerToClient.getModeNumber();
String unsuccessfulMsg = null;
boolean successful = true;
- boolean clientIsDurable = proxyId.isDurable();
+ boolean clientIsDurable = clientProxyMembershipID.isDurable();
if (logger.isDebugEnabled()) {
if (clientIsDurable) {
logger.debug("CacheClientNotifier: Attempting to register durable client: {}",
- proxyId.getDurableId());
+ clientProxyMembershipID.getDurableId());
} else {
logger.debug("CacheClientNotifier: Attempting to register non-durable client");
}
@@ -444,8 +274,8 @@ public class CacheClientNotifier {
byte endpointType = 0x00;
int queueSize = 0;
if (clientIsDurable) {
- if (l_proxy == null) {
- if (isTimedOut(proxyId)) {
+ if (cacheClientProxy == null) {
+ if (isTimedOut(clientProxyMembershipID)) {
queueSize = PoolImpl.PRIMARY_QUEUE_TIMED_OUT;
} else {
queueSize = PoolImpl.PRIMARY_QUEUE_NOT_AVAILABLE;
@@ -454,42 +284,45 @@ public class CacheClientNotifier {
if (logger.isDebugEnabled()) {
logger.debug(
"CacheClientNotifier: No proxy exists for durable client with id {}. It must be created.",
- proxyId.getDurableId());
+ clientProxyMembershipID.getDurableId());
}
- l_proxy =
- new CacheClientProxy(this, socket, proxyId, isPrimary, clientConflation, clientVersion,
+ cacheClientProxy =
+ new CacheClientProxy(this, socket, clientProxyMembershipID, isPrimary, clientConflation,
+ clientVersion,
acceptorId, notifyBySubscription, this.cache.getSecurityService(), subject);
- successful = this.initializeProxy(l_proxy);
+ successful = this.initializeProxy(cacheClientProxy);
} else {
- l_proxy.setSubject(subject);
- if (proxy.isPrimary()) {
+ cacheClientProxy.setSubject(subject);
+ if (cacheClientProxy.isPrimary()) {
endpointType = (byte) 2;
} else {
endpointType = (byte) 1;
}
- queueSize = proxy.getQueueSize();
+ queueSize = cacheClientProxy.getQueueSize();
// A proxy exists for this durable client. It must be reinitialized.
- if (l_proxy.isPaused()) {
+ if (cacheClientProxy.isPaused()) {
if (CacheClientProxy.testHook != null) {
CacheClientProxy.testHook.doTestHook("CLIENT_PRE_RECONNECT");
}
- if (l_proxy.lockDrain()) {
+ if (cacheClientProxy.lockDrain()) {
try {
if (logger.isDebugEnabled()) {
logger.debug(
"CacheClientNotifier: A proxy exists for durable client with id {}. This proxy will be reinitialized: {}",
- proxyId.getDurableId(), l_proxy);
+ clientProxyMembershipID.getDurableId(), cacheClientProxy);
}
this.statistics.incDurableReconnectionCount();
- l_proxy.getProxyID().updateDurableTimeout(proxyId.getDurableTimeout());
- l_proxy.reinitialize(socket, proxyId, this.getCache(), isPrimary, clientConflation,
+ cacheClientProxy.getProxyID()
+ .updateDurableTimeout(clientProxyMembershipID.getDurableTimeout());
+ cacheClientProxy.reinitialize(socket, clientProxyMembershipID, this.getCache(),
+ isPrimary, clientConflation,
clientVersion);
- l_proxy.setMarkerEnqueued(true);
+ cacheClientProxy.setMarkerEnqueued(true);
if (CacheClientProxy.testHook != null) {
CacheClientProxy.testHook.doTestHook("CLIENT_RECONNECTED");
}
} finally {
- l_proxy.unlockDrain();
+ cacheClientProxy.unlockDrain();
}
} else {
unsuccessfulMsg =
@@ -506,14 +339,14 @@ public class CacheClientNotifier {
unsuccessfulMsg =
String.format(
"The requested durable client has the same identifier ( %s ) as an existing durable client ( %s ). Duplicate durable clients are not allowed.",
- new Object[] {proxyId.getDurableId(), proxy});
+ new Object[] {clientProxyMembershipID.getDurableId(), cacheClientProxy});
logger.warn(unsuccessfulMsg);
// Set the unsuccessful response byte.
responseByte = Handshake.REPLY_EXCEPTION_DUPLICATE_DURABLE_CLIENT;
}
}
} else {
- CacheClientProxy staleClientProxy = this.getClientProxy(proxyId);
+ CacheClientProxy staleClientProxy = this.getClientProxy(clientProxyMembershipID);
if (staleClientProxy != null) {
// A proxy exists for this non-durable client. It must be closed.
if (logger.isDebugEnabled()) {
@@ -529,19 +362,20 @@ public class CacheClientNotifier {
} // non-null stale proxy
// Create the new proxy for this non-durable client
- l_proxy =
- new CacheClientProxy(this, socket, proxyId, isPrimary, clientConflation, clientVersion,
+ cacheClientProxy =
+ new CacheClientProxy(this, socket, clientProxyMembershipID, isPrimary, clientConflation,
+ clientVersion,
acceptorId, notifyBySubscription, this.cache.getSecurityService(), subject);
- successful = this.initializeProxy(l_proxy);
+ successful = this.initializeProxy(cacheClientProxy);
}
if (!successful) {
- l_proxy = null;
+ cacheClientProxy = null;
responseByte = Handshake.REPLY_REFUSED;
unsuccessfulMsg =
String.format(
"A previous connection attempt from this client is still being processed: %s",
- new Object[] {proxyId});
+ new Object[] {clientProxyMembershipID});
logger.warn(unsuccessfulMsg);
}
@@ -554,13 +388,14 @@ public class CacheClientNotifier {
DataOutputStream dos =
new DataOutputStream(new BufferedOutputStream(socket.getOutputStream()));
// write the message type, message length and the error message (if any)
- writeHandshakeMessage(dos, responseByte, unsuccessfulMsg, clientVersion, endpointType,
- queueSize);
+ socketMessageWriter.writeHandshakeMessage(dos, responseByte, unsuccessfulMsg, clientVersion,
+ endpointType, queueSize);
} catch (IOException ioe) {// remove the added proxy if we get IOException.
- if (l_proxy != null) {
- boolean keepProxy = l_proxy.close(false, false); // do not check for queue, just close it
+ if (cacheClientProxy != null) {
+ boolean keepProxy = cacheClientProxy.close(false, false); // do not check for queue, just
+ // close it
if (!keepProxy) {
- removeClientProxy(l_proxy);
+ removeClientProxy(cacheClientProxy);
}
}
throw ioe;
@@ -575,25 +410,41 @@ public class CacheClientNotifier {
// will ensure that the response byte is sent to the client before
// the marker message. If the client is durable, the message processor
// is not started until the clientReady message is received.
- if (!clientIsDurable && l_proxy != null
+ if (!clientIsDurable && cacheClientProxy != null
&& responseByte == CommunicationMode.SuccessfulServerToClient.getModeNumber()) {
// The startOrResumeMessageDispatcher tests if the proxy is a primary.
// If this is a secondary proxy, the dispatcher is not started.
// The false parameter signifies that a marker message has not already been
// processed. This will generate and send one.
- l_proxy.startOrResumeMessageDispatcher(false);
+ cacheClientProxy.startOrResumeMessageDispatcher(false);
}
if (responseByte == CommunicationMode.SuccessfulServerToClient.getModeNumber()) {
if (logger.isDebugEnabled()) {
- logger.debug("CacheClientNotifier: Successfully registered {}", l_proxy);
+ logger.debug("CacheClientNotifier: Successfully registered {}", cacheClientProxy);
}
} else {
logger.warn(
"CacheClientNotifier: Unsuccessfully registered client with identifier {} and response code {}",
- new Object[] {proxyId, responseByte});
+ new Object[] {clientProxyMembershipID, responseByte});
}
- return l_proxy;
+
+ performPostAuthorization(cacheClientProxy, clientProxyMembershipID, member,
+ sysProps,
+ subjectOrPrincipal);
+ }
+
+ private void handleAuthenticationException(final ClientProxyMembershipID clientProxyMembershipID,
+ final DataOutputStream dataOutputStream,
+ final Version clientVersion,
+ final GemFireSecurityException ex,
+ final byte replyExceptionAuthenticationFailed)
+ throws IOException {
+ securityLogWriter.warning(
+ String.format("An exception was thrown for client [%s]. %s",
+ clientProxyMembershipID, ex));
+ socketMessageWriter.writeException(dataOutputStream, replyExceptionAuthenticationFailed, ex,
+ clientVersion);
}
private boolean initializeProxy(CacheClientProxy l_proxy) throws IOException, CacheException {
@@ -734,7 +585,7 @@ public class CacheClientNotifier {
}
}
- private ClientUpdateMessageImpl constructClientMessage(InternalCacheEvent event) {
+ ClientUpdateMessageImpl constructClientMessage(InternalCacheEvent event) {
ClientUpdateMessageImpl clientMessage = null;
EnumListenerEvent operation = event.getEventType();
@@ -772,15 +623,11 @@ public class CacheClientNotifier {
}
private void singletonNotifyClients(InternalCacheEvent event, ClientUpdateMessage cmsg) {
- final boolean isDebugEnabled = logger.isDebugEnabled();
- final boolean isTraceEnabled = logger.isTraceEnabled();
-
FilterInfo filterInfo = event.getLocalFilterInfo();
- FilterProfile regionProfile = ((LocalRegion) event.getRegion()).getFilterProfile();
if (filterInfo != null) {
// if the routing was made using an old profile we need to recompute it
- if (isTraceEnabled) {
+ if (logger.isTraceEnabled()) {
logger.trace("Event isOriginRemote={}", event.isOriginRemote());
}
}
@@ -803,8 +650,62 @@ public class CacheClientNotifier {
return;
}
+ FilterProfile regionProfile = ((LocalRegion) event.getRegion()).getFilterProfile();
+
+ Set<ClientProxyMembershipID> filterClients =
+ getFilterClientIDs(event, regionProfile, filterInfo, clientMessage);
+
+ Conflatable conflatable;
+
+ if (clientMessage instanceof ClientTombstoneMessage) {
+ // bug #46832 - HAEventWrapper deserialization can't handle subclasses
+ // of ClientUpdateMessageImpl, so don't wrap them
+ conflatable = clientMessage;
+ // Remove clients older than 70 from the filterClients if the message is
+ // ClientTombstoneMessage. Fix for #46591.
+ Object[] objects = filterClients.toArray();
+ for (Object id : objects) {
+ CacheClientProxy ccp = getClientProxy((ClientProxyMembershipID) id, true);
+ if (ccp != null && ccp.getVersion().compareTo(Version.GFE_70) < 0) {
+ filterClients.remove(id);
+ }
+ }
+ } else {
+ HAEventWrapper wrapper = new HAEventWrapper(clientMessage);
+ wrapper.incrementPutInProgressCounter();
+
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "Initial increment PutInProgressCounter on HAEventWrapper with Event ID hash code: "
+ + wrapper.hashCode() + "; System ID hash code: "
+ + System.identityHashCode(wrapper) + "; Wrapper details: " + wrapper);
+ }
+
+ conflatable = wrapper;
+ }
+
+ registrationQueueManager.add(event, conflatable, filterClients, this);
+
+ singletonRouteClientMessage(conflatable, filterClients);
+
+ this.statistics.endEvent(startTime);
+
+ // Cleanup destroyed events in CQ result cache.
+ // While maintaining the CQ results key caching. the destroy event
+ // keys are marked as destroyed instead of removing them, this is
+ // to take care, arrival of duplicate events. The key marked as
+ // destroyed are removed after the event is placed in clients HAQueue.
+ if (filterInfo.filterProcessedLocally) {
+ removeDestroyTokensFromCqResultKeys(event, filterInfo);
+ }
+ }
+
+ Set<ClientProxyMembershipID> getFilterClientIDs(final InternalCacheEvent event,
+ final FilterProfile regionProfile,
+ final FilterInfo filterInfo,
+ final ClientUpdateMessageImpl clientMessage) {
// Holds the clientIds to which filter message needs to be sent.
- Set<ClientProxyMembershipID> filterClients = new HashSet();
+ Set<ClientProxyMembershipID> filterClients = new HashSet<>();
// Add CQ info.
if (filterInfo.getCQs() != null) {
@@ -818,7 +719,7 @@ public class CacheClientNotifier {
if (cq != null) {
ClientProxyMembershipID id = cq.getClientProxyId();
filterClients.add(id);
- if (isDebugEnabled) {
+ if (logger.isDebugEnabled()) {
logger.debug("Adding cq routing info to message for id: {} and cq: {}", id, cqName);
}
@@ -829,16 +730,12 @@ public class CacheClientNotifier {
// Add interestList info.
if (filterInfo.getInterestedClientsInv() != null) {
- Set<Object> rawIDs = regionProfile.getRealClientIDs(filterInfo.getInterestedClientsInv());
+ Set<Object> rawIDs =
+ regionProfile.getRealClientIDs(filterInfo.getInterestedClientsInv());
Set<ClientProxyMembershipID> ids = getProxyIDs(rawIDs, true);
- if (ids.remove(event.getContext())) { // don't send to member of origin
- CacheClientProxy ccp = getClientProxy(event.getContext());
- if (ccp != null) {
- ccp.getStatistics().incMessagesNotQueuedOriginator();
- }
- }
+ incMessagesNotQueuedOriginatorStat(event, ids);
if (!ids.isEmpty()) {
- if (isTraceEnabled) {
+ if (logger.isTraceEnabled()) {
logger.trace("adding invalidation routing to message for {}", ids);
}
clientMessage.addClientInterestList(ids, false);
@@ -848,14 +745,9 @@ public class CacheClientNotifier {
if (filterInfo.getInterestedClients() != null) {
Set<Object> rawIDs = regionProfile.getRealClientIDs(filterInfo.getInterestedClients());
Set<ClientProxyMembershipID> ids = getProxyIDs(rawIDs, true);
- if (ids.remove(event.getContext())) { // don't send to member of origin
- CacheClientProxy ccp = getClientProxy(event.getContext());
- if (ccp != null) {
- ccp.getStatistics().incMessagesNotQueuedOriginator();
- }
- }
+ incMessagesNotQueuedOriginatorStat(event, ids);
if (!ids.isEmpty()) {
- if (isTraceEnabled) {
+ if (logger.isTraceEnabled()) {
logger.trace("adding routing to message for {}", ids);
}
clientMessage.addClientInterestList(ids, true);
@@ -863,46 +755,28 @@ public class CacheClientNotifier {
}
}
- Conflatable conflatable = null;
-
- if (clientMessage instanceof ClientTombstoneMessage) {
- // bug #46832 - HAEventWrapper deserialization can't handle subclasses
- // of ClientUpdateMessageImpl, so don't wrap them
- conflatable = clientMessage;
- // Remove clients older than 70 from the filterClients if the message is
- // ClientTombstoneMessage. Fix for #46591.
- Object[] objects = filterClients.toArray();
- for (Object id : objects) {
- CacheClientProxy ccp = getClientProxy((ClientProxyMembershipID) id, true);
- if (ccp != null && ccp.getVersion().compareTo(Version.GFE_70) < 0) {
- filterClients.remove(id);
- }
- }
- } else {
- HAEventWrapper wrapper = new HAEventWrapper(clientMessage);
- wrapper.incrementPutInProgressCounter();
-
- if (logger.isDebugEnabled()) {
- logger.debug(
- "Initial increment PutInProgressCounter on HAEventWrapper with Event ID hash code: "
- + wrapper.hashCode() + "; System ID hash code: "
- + System.identityHashCode(wrapper) + "; Wrapper details: " + wrapper);
- }
+ return filterClients;
+ }
- conflatable = wrapper;
+ private boolean isClientPermitted(ClientRegistrationMetadata clientRegistrationMetadata,
+ ClientProxyMembershipID clientProxyMembershipID)
+ throws IOException {
+ if (getDenylistedClient().contains(clientProxyMembershipID)) {
+ Exception deniedException = new Exception("This client is denylisted by server");
+ socketMessageWriter.writeException(clientRegistrationMetadata.getDataOutputStream(),
+ Handshake.REPLY_INVALID, deniedException, clientRegistrationMetadata.getClientVersion());
+ return false;
}
+ return true;
+ }
- singletonRouteClientMessage(conflatable, filterClients);
-
- this.statistics.endEvent(startTime);
-
- // Cleanup destroyed events in CQ result cache.
- // While maintaining the CQ results key caching. the destroy event
- // keys are marked as destroyed instead of removing them, this is
- // to take care, arrival of duplicate events. The key marked as
- // destroyed are removed after the event is placed in clients HAQueue.
- if (filterInfo.filterProcessedLocally) {
- removeDestroyTokensFromCqResultKeys(event, filterInfo);
+ private void incMessagesNotQueuedOriginatorStat(final InternalCacheEvent event,
+ final Set<ClientProxyMembershipID> ids) {
+ if (ids.remove(event.getContext())) { // don't send to member of origin
+ CacheClientProxy ccp = getClientProxy(event.getContext());
+ if (ccp != null) {
+ ccp.getStatistics().incMessagesNotQueuedOriginator();
+ }
}
}
@@ -924,6 +798,57 @@ public class CacheClientNotifier {
}
}
+ private void performPostAuthorization(final CacheClientProxy proxy,
+ final ClientProxyMembershipID clientProxyMembershipID, final DistributedMember member,
+ final Properties sysProps, final Object subjectOrPrincipal)
+ throws ClassNotFoundException, NoSuchMethodException, IllegalAccessException,
+ InvocationTargetException {
+ if (proxy != null && subjectOrPrincipal != null) {
+ if (subjectOrPrincipal instanceof Principal) {
+ Principal principal = (Principal) subjectOrPrincipal;
+ if (securityLogWriter.fineEnabled()) {
+ securityLogWriter
+ .fine("CacheClientNotifier: successfully verified credentials for proxyID: "
+ + clientProxyMembershipID
+ + " having principal: " + principal.getName());
+ }
+
+ String postAuthzFactoryName = sysProps.getProperty(SECURITY_CLIENT_ACCESSOR_PP);
+ AccessControl authzCallback = null;
+ if (postAuthzFactoryName != null && postAuthzFactoryName.length() > 0) {
+ Method authzMethod = ClassLoadUtil.methodFromName(postAuthzFactoryName);
+ authzCallback = (AccessControl) authzMethod.invoke(null, (Object[]) null);
+ authzCallback.init(principal, member, this.getCache());
+ }
+ proxy.setPostAuthzCallback(authzCallback);
+ }
+ }
+ }
+
+ private Object getSubjectOrPrincipal(final ClientRegistrationMetadata clientRegistrationMetadata,
+ final DistributedMember member,
+ final DistributedSystem system, final String authenticator) {
+ final Object subjectOrPrincipal;
+
+ if (clientRegistrationMetadata.getClientCredentials() != null) {
+ if (securityLogWriter.fineEnabled()) {
+ securityLogWriter
+ .fine("CacheClientNotifier: verifying credentials for proxyID: "
+ + clientRegistrationMetadata.getClientProxyMembershipID());
+ }
+ subjectOrPrincipal = Handshake
+ .verifyCredentials(authenticator,
+ clientRegistrationMetadata.getClientCredentials(),
+ system.getSecurityProperties(),
+ this.logWriter, this.securityLogWriter, member,
+ this.cache.getSecurityService());
+ } else {
+ subjectOrPrincipal = null;
+ }
+
+ return subjectOrPrincipal;
+ }
+
/**
* delivers the given message to all proxies for routing. The message should already have client
* interest established, or override the isClientInterested method to implement its own routing
@@ -1844,7 +1769,7 @@ public class CacheClientNotifier {
*/
private CacheClientNotifier(InternalCache cache, CacheServerStats acceptorStats,
int maximumMessageCount, int messageTimeToLive, ConnectionListener listener,
- List overflowAttributesList, boolean isGatewayReceiver) {
+ boolean isGatewayReceiver) {
// Set the Cache
setCache(cache);
this.acceptorStats = acceptorStats;
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientProxy.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientProxy.java
index 8872ab8..acf21cb 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientProxy.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientProxy.java
@@ -29,7 +29,6 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
@@ -294,14 +293,6 @@ public class CacheClientProxy implements ClientSession {
/** acceptor's setting for notifyBySubscription */
private final boolean notifyBySubscription;
- /** To queue the events arriving during message dispatcher initialization */
- private volatile ConcurrentLinkedQueue<Conflatable> queuedEvents =
- new ConcurrentLinkedQueue<Conflatable>();
-
- private final Object queuedEventsSync = new Object();
-
- private volatile boolean messageDispatcherInit = false;
-
/**
* A counter that keeps track of how many task iterations that have occurred since the last ping
* or message. The {@linkplain CacheClientNotifier#scheduleClientPingTask ping task} increments
@@ -1591,37 +1582,6 @@ public class CacheClientProxy implements ClientSession {
}
if (clientMessage.needsNoAuthorizationCheck() || postDeliverAuthCheckPassed(clientMessage)) {
- // If dispatcher is getting initialized, add the event to temporary queue.
- if (this.messageDispatcherInit) {
- synchronized (this.queuedEventsSync) {
- if (this.messageDispatcherInit) { // Check to see value did not changed while getting the
- // synchronize lock.
- if (logger.isDebugEnabled()) {
- logger.debug(
- "Message dispatcher for proxy {} is getting initialized. Adding message to the queuedEvents.",
- this);
- }
-
- if (conflatable instanceof HAEventWrapper) {
- HAEventWrapper haEventWrapper = (HAEventWrapper) conflatable;
- haEventWrapper.incrementPutInProgressCounter();
- if (logger.isDebugEnabled()) {
- logger.debug(
- "Incremented PutInProgressCounter on HAEventWrapper with Event ID hash code: "
- + haEventWrapper.hashCode()
- + "; System ID hash code: "
- + System.identityHashCode(haEventWrapper) + "; Wrapper details: "
- + haEventWrapper);
- }
- }
-
- this.queuedEvents.add(conflatable);
-
- return;
- }
- }
- }
-
if (this._messageDispatcher != null) {
this._messageDispatcher.enqueueMessage(conflatable);
} else {
@@ -1730,7 +1690,6 @@ public class CacheClientProxy implements ClientSession {
*
*/
public void initializeMessageDispatcher() throws CacheException {
- this.messageDispatcherInit = true; // Initialization process.
try {
if (logger.isDebugEnabled()) {
logger.debug("{}: Initializing message dispatcher with capacity of {} entries", this,
@@ -1739,27 +1698,9 @@ public class CacheClientProxy implements ClientSession {
String name = "Client Message Dispatcher for " + getProxyID().getDistributedMember()
+ (isDurable() ? " (" + getDurableId() + ")" : "");
this._messageDispatcher = createMessageDispatcher(name);
-
- // Fix for 41375 - drain as many of the queued events
- // as we can without synchronization.
- if (logger.isDebugEnabled()) {
- logger.debug("{} draining {} events from init queue into intialized queue", this,
- this.queuedEvents.size());
- }
-
- drainQueuedEvents(false);
-
- // Now finish emptying the queue with synchronization to make
- // sure we don't miss any events.
- synchronized (this.queuedEventsSync) {
- drainQueuedEvents(true);
-
- this.messageDispatcherInit = false; // Done initialization.
- }
- } finally {
- if (this.messageDispatcherInit) { // If its not successfully completed.
- this._statistics.close();
- }
+ } catch (final Exception ex) {
+ this._statistics.close();
+ throw ex;
}
}
@@ -1767,29 +1708,6 @@ public class CacheClientProxy implements ClientSession {
return new MessageDispatcher(this, name);
}
- private void drainQueuedEvents(boolean withSynchronization) {
- Conflatable nextEvent;
- while ((nextEvent = queuedEvents.poll()) != null) {
- if (logger.isDebugEnabled()) {
- if (nextEvent instanceof HAEventWrapper) {
- logger.debug(
- "Draining events queued during message dispatcher initialization "
- + (withSynchronization ? "with" : "without")
- + " synchronization. Event ID hash code: "
- + nextEvent.hashCode()
- + "; System ID hash code: " + System.identityHashCode(nextEvent)
- + "; Wrapper details: " + nextEvent);
- }
- }
-
- this._messageDispatcher.enqueueMessage(nextEvent);
-
- if (nextEvent instanceof HAEventWrapper) {
- ((HAEventWrapper) nextEvent).decrementPutInProgressCounter();
- }
- }
- }
-
protected void startOrResumeMessageDispatcher(boolean processedMarker) {
// Only start or resume the dispatcher if it is Primary
if (this.isPrimary) {
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientRegistrationEventQueueManager.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientRegistrationEventQueueManager.java
new file mode 100644
index 0000000..122a192
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientRegistrationEventQueueManager.java
@@ -0,0 +1,279 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.geode.internal.cache.tier.sockets;
+
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.ReadWriteLock;
+
+import org.apache.logging.log4j.Logger;
+
+import org.apache.geode.internal.cache.Conflatable;
+import org.apache.geode.internal.cache.FilterProfile;
+import org.apache.geode.internal.cache.FilterRoutingInfo;
+import org.apache.geode.internal.cache.InternalCacheEvent;
+import org.apache.geode.internal.cache.LocalRegion;
+import org.apache.geode.internal.logging.LogService;
+
+/**
+ * This class is responsible for queueing events for clients while they are registering. Once the
+ * client is completely registered (filter info retrieved and GII complete), we will drain the
+ * client's queued events and deliver them to the cache client proxy if necessary.
+ */
+class ClientRegistrationEventQueueManager {
+ private static final Logger logger = LogService.getLogger();
+ private final Map<ClientProxyMembershipID, ClientRegistrationEventQueue> registeringProxyEventQueues =
+ new ConcurrentHashMap<>();
+
+ void add(final InternalCacheEvent event,
+ final Conflatable conflatable,
+ final Set<ClientProxyMembershipID> originalFilterClientIDs,
+ final CacheClientNotifier cacheClientNotifier) {
+ if (registeringProxyEventQueues.isEmpty())
+ return;
+
+ ClientRegistrationEvent clientRegistrationEvent =
+ new ClientRegistrationEvent(event, conflatable);
+
+ for (final Map.Entry<ClientProxyMembershipID, ClientRegistrationEventQueue> eventsReceivedWhileRegisteringClient : registeringProxyEventQueues
+ .entrySet()) {
+ ClientProxyMembershipID clientProxyMembershipID =
+ eventsReceivedWhileRegisteringClient.getKey();
+ ClientRegistrationEventQueue registrationEventQueue =
+ eventsReceivedWhileRegisteringClient.getValue();
+
+ registrationEventQueue.lockForPutting();
+ try {
+ // If this is an HAEventWrapper we need to increment the PutInProgress counter so
+ // that the contents of the contents of the HAEventWrapper are preserved when the
+ // event is drained and processed. See incrementPutInProgressCounter() and
+ // decrementPutInProgressCounter() for more details.
+ if (conflatable instanceof HAEventWrapper) {
+ ((HAEventWrapper) conflatable).incrementPutInProgressCounter();
+ }
+
+ // After taking out the lock, we need to determine if the client is still actually
+ // registering since there is a small race where it may have finished registering
+ // after we pulled the queue out of the registeringProxyEventQueues collection
+ if (registeringProxyEventQueues.containsKey(clientProxyMembershipID)) {
+ registrationEventQueue.add(clientRegistrationEvent);
+
+ // Because this event will be processed and sent when it is drained out of the temporary
+ // client registration queue, we do not need to send it to the client at this time.
+ // We can prevent the event from being sent to the registering client at this time
+ // by removing its client proxy membership ID from the filter clients collection,
+ // if it exists.
+ originalFilterClientIDs.remove(clientProxyMembershipID);
+ } else {
+ // The client is no longer registering, but we want to reprocess the filter info on
+ // this event and potentially deliver the conflatable to handle the edge case where
+ // the original client filter IDs generated in the "normal" put processing path did
+ // not include the registering client because the filter info was not yet available.
+ processEventAndDeliverConflatable(clientProxyMembershipID, cacheClientNotifier, event,
+ conflatable, originalFilterClientIDs);
+ }
+ } finally {
+ registrationEventQueue.unlockForPutting();
+ }
+ }
+ }
+
+ void drain(final ClientProxyMembershipID clientProxyMembershipID,
+ final CacheClientNotifier cacheClientNotifier) {
+ // As an optimization, we drain as many events from the queue as we can
+ // before taking out a lock to drain the remaining events
+ if (logger.isDebugEnabled()) {
+ logger.debug("Draining events from registration queue for client proxy "
+ + clientProxyMembershipID
+ + " without synchronization");
+ }
+
+ drainEventsReceivedWhileRegisteringClient(clientProxyMembershipID, cacheClientNotifier);
+
+ ClientRegistrationEventQueue registrationEventQueue =
+ registeringProxyEventQueues.get(clientProxyMembershipID);
+
+ registrationEventQueue.lockForDraining();
+ try {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Draining remaining events from registration queue for client proxy "
+ + clientProxyMembershipID + " with synchronization");
+ }
+
+ drainEventsReceivedWhileRegisteringClient(clientProxyMembershipID, cacheClientNotifier);
+
+ registeringProxyEventQueues.remove(clientProxyMembershipID);
+ } finally {
+ registrationEventQueue.unlockForDraining();
+ }
+ }
+
+ private void drainEventsReceivedWhileRegisteringClient(final ClientProxyMembershipID proxyID,
+ final CacheClientNotifier cacheClientNotifier) {
+ ClientRegistrationEvent queuedEvent;
+ ClientRegistrationEventQueue registrationEventQueue = registeringProxyEventQueues.get(proxyID);
+
+ while ((queuedEvent = registrationEventQueue.poll()) != null) {
+ InternalCacheEvent internalCacheEvent = queuedEvent.internalCacheEvent;
+ Conflatable conflatable = queuedEvent.conflatable;
+ processEventAndDeliverConflatable(proxyID, cacheClientNotifier, internalCacheEvent,
+ conflatable, null);
+ }
+ }
+
+ public ClientRegistrationEventQueue create(
+ final ClientProxyMembershipID clientProxyMembershipID,
+ final Queue<ClientRegistrationEvent> eventQueue,
+ final ReadWriteLock putDrainLock) {
+ final ClientRegistrationEventQueue clientRegistrationEventQueue =
+ new ClientRegistrationEventQueue(eventQueue,
+ putDrainLock);
+ registeringProxyEventQueues.put(clientProxyMembershipID,
+ clientRegistrationEventQueue);
+ return clientRegistrationEventQueue;
+ }
+
+ class ClientRegistrationEventQueue {
+ Queue<ClientRegistrationEvent> eventQueue;
+ ReadWriteLock readWriteLock;
+
+ ClientRegistrationEventQueue(
+ final Queue<ClientRegistrationEvent> eventQueue, final ReadWriteLock readWriteLock) {
+ this.eventQueue = eventQueue;
+ this.readWriteLock = readWriteLock;
+ }
+
+ boolean isEmpty() {
+ return eventQueue.isEmpty();
+ }
+
+ private void add(final ClientRegistrationEvent clientRegistrationEvent) {
+ eventQueue.add(clientRegistrationEvent);
+ }
+
+ public ClientRegistrationEvent poll() {
+ return eventQueue.poll();
+ }
+
+ private void lockForDraining() {
+ readWriteLock.writeLock().lock();
+ }
+
+ private void unlockForDraining() {
+ readWriteLock.writeLock().unlock();
+ }
+
+ private void lockForPutting() {
+ readWriteLock.readLock().lock();
+ }
+
+ private void unlockForPutting() {
+ readWriteLock.readLock().unlock();
+ }
+ }
+
+ private void processEventAndDeliverConflatable(final ClientProxyMembershipID proxyID,
+ final CacheClientNotifier cacheClientNotifier,
+ final InternalCacheEvent internalCacheEvent,
+ final Conflatable conflatable,
+ final Set<ClientProxyMembershipID> originalFilterClientIDs) {
+ // The first step is to repopulate the filter info for the event to determine if
+ // the client which was registering has a matching CQ or has registered interest
+ // in the key for this event. We need to get the filter profile, filter routing info,
+ // and local filter info in order to do so. If any of these are null, then there is
+ // no need to proceed as the client is not interested.
+ FilterProfile filterProfile =
+ ((LocalRegion) internalCacheEvent.getRegion()).getFilterProfile();
+
+ if (filterProfile != null) {
+ FilterRoutingInfo filterRoutingInfo =
+ filterProfile.getFilterRoutingInfoPart2(null, internalCacheEvent);
+
+ if (filterRoutingInfo != null) {
+ FilterRoutingInfo.FilterInfo filterInfo = filterRoutingInfo.getLocalFilterInfo();
+
+ if (filterInfo != null) {
+ ClientUpdateMessageImpl clientUpdateMessage = conflatable instanceof HAEventWrapper
+ ? (ClientUpdateMessageImpl) ((HAEventWrapper) conflatable).getClientUpdateMessage()
+ : (ClientUpdateMessageImpl) conflatable;
+
+ internalCacheEvent.setLocalFilterInfo(filterInfo);
+
+ Set<ClientProxyMembershipID> newFilterClientIDs =
+ cacheClientNotifier.getFilterClientIDs(internalCacheEvent, filterProfile,
+ filterInfo,
+ clientUpdateMessage);
+
+ if (eventNotInOriginalFilterClientIDs(proxyID, newFilterClientIDs,
+ originalFilterClientIDs)) {
+ CacheClientProxy cacheClientProxy = cacheClientNotifier.getClientProxy(proxyID);
+
+ if (eventShouldBeDelivered(proxyID, newFilterClientIDs, cacheClientProxy)) {
+ cacheClientProxy.deliverMessage(conflatable);
+ }
+ }
+ }
+ }
+ }
+
+ // Once we have processed the conflatable, if it is an HAEventWrapper we can
+ // decrement the PutInProgress counter, allowing the ClientUpdateMessage to be
+ // set to null. See decrementPutInProgressCounter() for more details.
+ if (conflatable instanceof HAEventWrapper) {
+ ((HAEventWrapper) conflatable).decrementPutInProgressCounter();
+ }
+ }
+
+ private boolean eventShouldBeDelivered(final ClientProxyMembershipID proxyID,
+ final Set<ClientProxyMembershipID> filterClientIDs,
+ final CacheClientProxy cacheClientProxy) {
+ return filterClientIDs.contains(proxyID) && cacheClientProxy != null;
+ }
+
+ /*
+ * This is to handle the edge case where the original filter client IDs
+ * calculated by "normal" put processing did not include the registering client
+ * because the filter info had not been received yet, but we now found that the client
+ * is interested in the event so we should deliver it.
+ */
+ private boolean eventNotInOriginalFilterClientIDs(final ClientProxyMembershipID proxyID,
+ final Set<ClientProxyMembershipID> newFilterClientIDs,
+ final Set<ClientProxyMembershipID> originalFilterClientIDs) {
+ return originalFilterClientIDs == null
+ || (!originalFilterClientIDs.contains(proxyID) && newFilterClientIDs.contains(proxyID));
+ }
+
+ /**
+ * Represents a conflatable and event processed while a client was registering.
+ * This needs to be queued and processed after registration is complete. The conflatable
+ * is what we will actually be delivering to the MessageDispatcher (and thereby adding
+ * to the HARegionQueue). The internal cache event is required to rehydrate the filter
+ * info and determine if the client which was registering does have a CQ that matches or
+ * has registered interest in the key.
+ */
+ private class ClientRegistrationEvent {
+ private final Conflatable conflatable;
+ private final InternalCacheEvent internalCacheEvent;
+
+ ClientRegistrationEvent(final InternalCacheEvent internalCacheEvent,
+ final Conflatable conflatable) {
+ this.conflatable = conflatable;
+ this.internalCacheEvent = internalCacheEvent;
+ }
+ }
+}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientRegistrationMetadata.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientRegistrationMetadata.java
new file mode 100644
index 0000000..3c83905
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientRegistrationMetadata.java
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.geode.internal.cache.tier.sockets;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.net.Socket;
+import java.net.SocketAddress;
+import java.util.Properties;
+
+import org.apache.logging.log4j.Logger;
+
+import org.apache.geode.cache.UnsupportedVersionException;
+import org.apache.geode.internal.Version;
+import org.apache.geode.internal.VersionedDataInputStream;
+import org.apache.geode.internal.VersionedDataOutputStream;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.tier.CommunicationMode;
+import org.apache.geode.internal.logging.LogService;
+
+class ClientRegistrationMetadata {
+ private static final Logger logger = LogService.getLogger();
+ private final InternalCache cache;
+ private final Socket socket;
+ private final SocketMessageWriter socketMessageWriter;
+ private ClientProxyMembershipID clientProxyMembershipID;
+ private byte clientConflation;
+ private Properties clientCredentials;
+ private Version clientVersion;
+ private DataInputStream dataInputStream;
+ private DataOutputStream dataOutputStream;
+
+ ClientRegistrationMetadata(final InternalCache cache, final Socket socket) {
+ this.cache = cache;
+ this.socket = socket;
+ this.socketMessageWriter = new SocketMessageWriter();
+ }
+
+ boolean initialize() throws IOException {
+ DataInputStream unversionedDataInputStream = new DataInputStream(socket.getInputStream());
+ DataOutputStream unversionedDataOutputStream = new DataOutputStream(socket.getOutputStream());
+
+ if (getAndValidateClientVersion(socket, unversionedDataInputStream,
+ unversionedDataOutputStream)) {
+ if (oldClientRequiresVersionedStreams(clientVersion)) {
+ dataInputStream = new VersionedDataInputStream(unversionedDataInputStream, clientVersion);
+ dataOutputStream =
+ new VersionedDataOutputStream(unversionedDataOutputStream, clientVersion);
+ } else {
+ dataInputStream = unversionedDataInputStream;
+ dataOutputStream = unversionedDataOutputStream;
+ }
+
+ // Read and ignore the reply code. This is used on the client to server
+ // handshake.
+ dataInputStream.readByte(); // replyCode
+
+ // Read the ports and throw them away. We no longer need them
+ int numberOfPorts = dataInputStream.readInt();
+ for (int i = 0; i < numberOfPorts; i++) {
+ dataInputStream.readInt();
+ }
+
+ getAndValidateClientProxyMembershipID();
+
+ if (getAndValidateClientConflation()) {
+ clientCredentials =
+ Handshake.readCredentials(dataInputStream, dataOutputStream,
+ cache.getDistributedSystem(), cache.getSecurityService());
+
+ return true;
+ }
+ }
+
+ return false;
+ }
+
+ ClientProxyMembershipID getClientProxyMembershipID() {
+ return clientProxyMembershipID;
+ }
+
+ byte getClientConflation() {
+ return clientConflation;
+ }
+
+ Properties getClientCredentials() {
+ return clientCredentials;
+ }
+
+ Version getClientVersion() {
+ return clientVersion;
+ }
+
+ DataOutputStream getDataOutputStream() {
+ return dataOutputStream;
+ }
+
+ private boolean getAndValidateClientVersion(final Socket socket,
+ final DataInputStream dataInputStream, final DataOutputStream dataOutputStream)
+ throws IOException {
+ short clientVersionOrdinal = Version.readOrdinal(dataInputStream);
+
+ try {
+ clientVersion = Version.fromOrdinal(clientVersionOrdinal, true);
+ if (isVersionOlderThan57(clientVersion)) {
+ throw new IOException(new UnsupportedVersionException(clientVersionOrdinal));
+ }
+ if (logger.isDebugEnabled()) {
+ logger.debug("{}: Registering client with version: {}", this, clientVersion);
+ }
+ } catch (UnsupportedVersionException e) {
+ UnsupportedVersionException unsupportedVersionException = e;
+ SocketAddress socketAddress = socket.getRemoteSocketAddress();
+
+ if (socketAddress != null) {
+ String sInfo = " Client: " + socketAddress.toString() + ".";
+ unsupportedVersionException = new UnsupportedVersionException(e.getMessage() + sInfo);
+ }
+
+ logger.warn(
+ "CacheClientNotifier: Registering client version is unsupported. Error details: ",
+ unsupportedVersionException);
+
+ socketMessageWriter.writeException(dataOutputStream,
+ CommunicationMode.UnsuccessfulServerToClient.getModeNumber(),
+ unsupportedVersionException, null);
+
+ return false;
+ }
+
+ return true;
+ }
+
+ private boolean doesClientSupportExtractOverrides() {
+ return clientVersion.compareTo(Version.GFE_603) >= 0;
+ }
+
+ private boolean oldClientRequiresVersionedStreams(final Version clientVersion) {
+ return Version.CURRENT.compareTo(clientVersion) > 0;
+ }
+
+ private boolean isVersionOlderThan57(final Version clientVersion) {
+ return Version.GFE_57.compareTo(clientVersion) > 0;
+ }
+
+ private void getAndValidateClientProxyMembershipID()
+ throws IOException {
+ try {
+ clientProxyMembershipID = ClientProxyMembershipID.readCanonicalized(dataInputStream);
+ } catch (ClassNotFoundException ex) {
+ throw new IOException(ex);
+ }
+ }
+
+ private boolean getAndValidateClientConflation()
+ throws IOException {
+ if (doesClientSupportExtractOverrides()) {
+ byte[] overrides =
+ Handshake.extractOverrides(new byte[] {(byte) dataInputStream.read()});
+ clientConflation = overrides[0];
+ } else {
+ clientConflation = (byte) dataInputStream.read();
+ }
+
+ switch (clientConflation) {
+ case Handshake.CONFLATION_DEFAULT:
+ case Handshake.CONFLATION_OFF:
+ case Handshake.CONFLATION_ON:
+ break;
+ default:
+ socketMessageWriter.writeException(dataOutputStream, Handshake.REPLY_INVALID,
+ new IllegalArgumentException("Invalid conflation byte"), clientVersion);
+
+ return false;
+ }
+
+ return true;
+ }
+}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/SocketMessageWriter.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/SocketMessageWriter.java
new file mode 100644
index 0000000..095d8ff
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/SocketMessageWriter.java
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.geode.internal.cache.tier.sockets;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+
+import org.apache.geode.DataSerializer;
+import org.apache.geode.Instantiator;
+import org.apache.geode.distributed.internal.DistributionConfig;
+import org.apache.geode.internal.InternalDataSerializer;
+import org.apache.geode.internal.InternalInstantiator;
+import org.apache.geode.internal.Version;
+
+public class SocketMessageWriter {
+ private static final int CLIENT_PING_TASK_PERIOD =
+ Integer.getInteger(DistributionConfig.GEMFIRE_PREFIX + "serverToClientPingPeriod", 60000);
+
+ public void writeHandshakeMessage(DataOutputStream dos, byte type, String p_msg,
+ Version clientVersion, byte endpointType, int queueSize) throws IOException {
+ String msg = p_msg;
+
+ // write the message type
+ dos.writeByte(type);
+
+ dos.writeByte(endpointType);
+ dos.writeInt(queueSize);
+
+ if (msg == null) {
+ msg = "";
+ }
+ dos.writeUTF(msg);
+ if (clientVersion != null && clientVersion.compareTo(Version.GFE_61) >= 0) {
+ // get all the instantiators.
+ Instantiator[] instantiators = InternalInstantiator.getInstantiators();
+ HashMap instantiatorMap = new HashMap();
+ if (instantiators != null && instantiators.length > 0) {
+ for (Instantiator instantiator : instantiators) {
+ ArrayList instantiatorAttributes = new ArrayList();
+ instantiatorAttributes.add(instantiator.getClass().toString().substring(6));
+ instantiatorAttributes.add(instantiator.getInstantiatedClass().toString().substring(6));
+ instantiatorMap.put(instantiator.getId(), instantiatorAttributes);
+ }
+ }
+ DataSerializer.writeHashMap(instantiatorMap, dos);
+
+ // get all the dataserializers.
+ DataSerializer[] dataSerializers = InternalDataSerializer.getSerializers();
+ HashMap<Integer, ArrayList<String>> dsToSupportedClasses =
+ new HashMap<Integer, ArrayList<String>>();
+ HashMap<Integer, String> dataSerializersMap = new HashMap<Integer, String>();
+ if (dataSerializers != null && dataSerializers.length > 0) {
+ for (DataSerializer dataSerializer : dataSerializers) {
+ dataSerializersMap.put(dataSerializer.getId(),
+ dataSerializer.getClass().toString().substring(6));
+ if (clientVersion.compareTo(Version.GFE_6516) >= 0) {
+ ArrayList<String> supportedClassNames = new ArrayList<String>();
+ for (Class clazz : dataSerializer.getSupportedClasses()) {
+ supportedClassNames.add(clazz.getName());
+ }
+ dsToSupportedClasses.put(dataSerializer.getId(), supportedClassNames);
+ }
+ }
+ }
+ DataSerializer.writeHashMap(dataSerializersMap, dos);
+ if (clientVersion.compareTo(Version.GFE_6516) >= 0) {
+ DataSerializer.writeHashMap(dsToSupportedClasses, dos);
+ }
+ if (clientVersion.compareTo(Version.GEODE_1_5_0) >= 0) {
+ dos.writeInt(CLIENT_PING_TASK_PERIOD);
+ }
+ }
+ dos.flush();
+ }
+
+ /**
+ * Writes an exception message to the socket
+ *
+ * @param dos the <code>DataOutputStream</code> to use for writing the message
+ * @param type a byte representing the exception type
+ * @param ex the exception to be written; should not be null
+ */
+ public void writeException(DataOutputStream dos, byte type, Exception ex, Version clientVersion)
+ throws IOException {
+ writeHandshakeMessage(dos, type, ex.toString(), clientVersion, (byte) 0x00, 0);
+ }
+}
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifierTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifierTest.java
new file mode 100644
index 0000000..5cbdfe4
--- /dev/null
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifierTest.java
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.geode.internal.cache.tier.sockets;
+
+import static org.mockito.ArgumentMatchers.isA;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.net.Socket;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.junit.Test;
+
+import org.apache.geode.cache.query.internal.cq.ServerCQ;
+import org.apache.geode.internal.cache.DistributedRegion;
+import org.apache.geode.internal.cache.FilterProfile;
+import org.apache.geode.internal.cache.FilterRoutingInfo;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.InternalCacheEvent;
+import org.apache.geode.test.fake.Fakes;
+
+public class CacheClientNotifierTest {
+ @Test
+ public void eventsInClientRegistrationQueueAreSentToClientAfterRegistrationIsComplete()
+ throws IOException, ClassNotFoundException, NoSuchMethodException, IllegalAccessException,
+ InvocationTargetException, InterruptedException, ExecutionException {
+ InternalCache internalCache = Fakes.cache();
+ CacheServerStats cacheServerStats = mock(CacheServerStats.class);
+ Socket socket = mock(Socket.class);
+ ConnectionListener connectionListener = mock(ConnectionListener.class);
+ ClientProxyMembershipID clientProxyMembershipID = mock(ClientProxyMembershipID.class);
+ CacheClientProxy cacheClientProxy = mock(CacheClientProxy.class);
+ ClientUpdateMessageImpl clientUpdateMessage = mock(ClientUpdateMessageImpl.class);
+ ClientRegistrationMetadata clientRegistrationMetadata = mock(ClientRegistrationMetadata.class);
+ when(clientRegistrationMetadata.getClientProxyMembershipID()).thenReturn(
+ clientProxyMembershipID);
+
+ CacheClientNotifier cacheClientNotifier = CacheClientNotifier.getInstance(internalCache,
+ cacheServerStats, 0, 0, connectionListener, null, false);
+ final CacheClientNotifier cacheClientNotifierSpy = spy(cacheClientNotifier);
+
+ CountDownLatch waitForEventDispatchCountdownLatch = new CountDownLatch(1);
+ CountDownLatch waitForRegistrationCountdownLatch = new CountDownLatch(1);
+ ExecutorService registerAndNotifyExecutor = Executors.newFixedThreadPool(2);
+
+ try {
+ // We stub out the CacheClientNotifier.registerClientInternal() to do some "work" until
+ // a new event is received and queued, as triggered by the waitForEventDispatchCountdownLatch
+ doAnswer((i) -> {
+ when(cacheClientProxy.getProxyID()).thenReturn(clientProxyMembershipID);
+ cacheClientNotifierSpy.addClientProxy(cacheClientProxy);
+ waitForRegistrationCountdownLatch.countDown();
+ waitForEventDispatchCountdownLatch.await();
+ return null;
+ }).when(cacheClientNotifierSpy).registerClientInternal(clientRegistrationMetadata, socket,
+ false, 0, true);
+ List<Callable<Void>> registerAndNotifyTasks = new ArrayList<>();
+
+ // In one thread, we register the new client which should create the temporary client
+ // registration event queue. Events will be passed to that queue while registration is
+ // underway. Once registration is complete, the queue is drained and the event is processed
+ // as normal.
+ registerAndNotifyTasks.add(() -> {
+ cacheClientNotifierSpy.registerClient(clientRegistrationMetadata, socket, false, 0, true);
+ return null;
+ });
+
+ // In a second thread, we mock the arrival of a new event. We want to ensure this event
+ // goes into the temporary client registration event queue. To do that, we wait on the
+ // waitForRegistrationCountdownLatch until registration is underway and the temp queue is
+ // created. Once it is, we process the event and notify clients, which should add the event
+ // to the temp queue. Finally, we resume registration and after it is complete, we verify
+ // that the event was drained, processed, and "delivered" (note message delivery is mocked
+ // and results in a no-op).
+ registerAndNotifyTasks.add(() -> {
+ try {
+ waitForRegistrationCountdownLatch.await();
+
+ InternalCacheEvent internalCacheEvent =
+ createMockInternalCacheEvent(clientProxyMembershipID, clientUpdateMessage,
+ cacheClientNotifierSpy);
+
+ CacheClientNotifier.notifyClients(internalCacheEvent, clientUpdateMessage);
+ } finally {
+ // Ensure we always countdown so if the test fails it won't hang due to
+ // awaiting on this countdown latch.
+ waitForEventDispatchCountdownLatch.countDown();
+ }
+ return null;
+ });
+
+ final List<Future<Void>> futures =
+ registerAndNotifyExecutor.invokeAll(registerAndNotifyTasks);
+
+ for (final Future future : futures) {
+ future.get();
+ }
+ } finally {
+ // To prevent not cleaning up test resources in case an unexpected exception occurs
+ waitForEventDispatchCountdownLatch.countDown();
+ waitForRegistrationCountdownLatch.countDown();
+ registerAndNotifyExecutor.shutdownNow();
+ cacheClientNotifier.shutdown(0);
+ }
+
+ verify(cacheClientProxy, times(1)).deliverMessage(isA(HAEventWrapper.class));
+ }
+
+ private InternalCacheEvent createMockInternalCacheEvent(
+ final ClientProxyMembershipID clientProxyMembershipID,
+ final ClientUpdateMessageImpl clientUpdateMessage,
+ final CacheClientNotifier cacheClientNotifierSpy) {
+ InternalCacheEvent internalCacheEvent = mock(InternalCacheEvent.class);
+
+ DistributedRegion region = mock(DistributedRegion.class);
+ when(internalCacheEvent.getRegion()).thenReturn(region);
+
+ FilterRoutingInfo.FilterInfo filterInfo = mock(FilterRoutingInfo.FilterInfo.class);
+ final Long cqId = 0L;
+ final String cqName = "testCQ";
+
+ HashMap cqs = new HashMap<Long, Integer>() {
+ {
+ put(cqId, 123);
+ }
+ };
+ when(filterInfo.getCQs()).thenReturn(cqs);
+ when(internalCacheEvent.getLocalFilterInfo()).thenReturn(
+ filterInfo);
+
+ FilterProfile filterProfile = mock(FilterProfile.class);
+ when(filterProfile.getRealCqID(cqId)).thenReturn(cqName);
+
+ ServerCQ serverCQ = mock(ServerCQ.class);
+ when(serverCQ.getClientProxyId()).thenReturn(clientProxyMembershipID);
+ when(filterProfile.getCq(cqName)).thenReturn(serverCQ);
+ when(region.getFilterProfile()).thenReturn(filterProfile);
+
+ FilterRoutingInfo filterRoutingInfo = mock(FilterRoutingInfo.class);
+ when(filterRoutingInfo.getLocalFilterInfo()).thenReturn(filterInfo);
+ when(filterProfile.getFilterRoutingInfoPart2(null, internalCacheEvent))
+ .thenReturn(filterRoutingInfo);
+ doReturn(clientUpdateMessage).when(cacheClientNotifierSpy)
+ .constructClientMessage(internalCacheEvent);
+
+ return internalCacheEvent;
+ }
+}
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/ClientRegistrationEventQueueManagerTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/ClientRegistrationEventQueueManagerTest.java
new file mode 100644
index 0000000..801d178
--- /dev/null
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/ClientRegistrationEventQueueManagerTest.java
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.geode.internal.cache.tier.sockets;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import org.junit.Test;
+
+import org.apache.geode.internal.cache.Conflatable;
+import org.apache.geode.internal.cache.FilterProfile;
+import org.apache.geode.internal.cache.FilterRoutingInfo;
+import org.apache.geode.internal.cache.InternalCacheEvent;
+import org.apache.geode.internal.cache.LocalRegion;
+
+public class ClientRegistrationEventQueueManagerTest {
+ @Test
+ public void messageDeliveredAfterRegisteringOnDrainIfNewFilterIDsIncludesClient()
+ throws ExecutionException, InterruptedException {
+ ClientRegistrationEventQueueManager clientRegistrationEventQueueManager =
+ new ClientRegistrationEventQueueManager();
+
+ ClientProxyMembershipID clientProxyMembershipID = mock(ClientProxyMembershipID.class);
+
+ ReadWriteLock mockPutDrainLock = mock(ReadWriteLock.class);
+ ReadWriteLock actualPutDrainLock = new ReentrantReadWriteLock();
+
+ CountDownLatch waitForDrainAndRemoveLatch = new CountDownLatch(1);
+ CountDownLatch waitForAddInProgressLatch = new CountDownLatch(1);
+
+ when(mockPutDrainLock.readLock())
+ .thenAnswer(i -> {
+ waitForAddInProgressLatch.countDown();
+ waitForDrainAndRemoveLatch.await();
+ return actualPutDrainLock.readLock();
+ });
+
+ when(mockPutDrainLock.writeLock())
+ .thenAnswer(i -> {
+ waitForAddInProgressLatch.await();
+ waitForDrainAndRemoveLatch.countDown();
+ return actualPutDrainLock.writeLock();
+ });
+
+ clientRegistrationEventQueueManager.create(clientProxyMembershipID,
+ new ConcurrentLinkedQueue<>(), mockPutDrainLock);
+
+ InternalCacheEvent internalCacheEvent = mock(InternalCacheEvent.class);
+ LocalRegion localRegion = mock(LocalRegion.class);
+ FilterProfile filterProfile = mock(FilterProfile.class);
+ FilterRoutingInfo filterRoutingInfo = mock(FilterRoutingInfo.class);
+ FilterRoutingInfo.FilterInfo filterInfo = mock(FilterRoutingInfo.FilterInfo.class);
+
+ when(filterRoutingInfo.getLocalFilterInfo()).thenReturn(
+ filterInfo);
+ when(filterProfile.getFilterRoutingInfoPart2(null, internalCacheEvent))
+ .thenReturn(filterRoutingInfo);
+ when(localRegion.getFilterProfile()).thenReturn(filterProfile);
+ when(internalCacheEvent.getRegion()).thenReturn(localRegion);
+
+ ClientUpdateMessageImpl clientUpdateMessage = mock(ClientUpdateMessageImpl.class);
+
+ CacheClientNotifier cacheClientNotifier = mock(CacheClientNotifier.class);
+ Set<ClientProxyMembershipID> recalculatedFilterClientIDs = new HashSet<>();
+ recalculatedFilterClientIDs.add(clientProxyMembershipID);
+ when(cacheClientNotifier.getFilterClientIDs(internalCacheEvent, filterProfile, filterInfo,
+ clientUpdateMessage))
+ .thenReturn(recalculatedFilterClientIDs);
+ CacheClientProxy cacheClientProxy = mock(CacheClientProxy.class);
+ when(cacheClientNotifier.getClientProxy(clientProxyMembershipID)).thenReturn(cacheClientProxy);
+
+ // Create empty filter client IDs produced by the "normal" put processing path, so we can test
+ // that the event is still delivered if the client finished registering and needs the event.
+ Set<ClientProxyMembershipID> normalPutFilterClientIDs = new HashSet<>();
+ CompletableFuture<Void> addEventsToQueueTask = CompletableFuture.runAsync(() -> {
+ // In thread one, we add and event to the queue
+ clientRegistrationEventQueueManager
+ .add(internalCacheEvent, clientUpdateMessage, normalPutFilterClientIDs,
+ cacheClientNotifier);
+ });
+
+ CompletableFuture<Void> drainEventsFromQueueTask = CompletableFuture.runAsync(() -> {
+ // In thread two, we drain the event from the queue
+ clientRegistrationEventQueueManager.drain(clientProxyMembershipID, cacheClientNotifier);
+ });
+
+ CompletableFuture.allOf(addEventsToQueueTask, drainEventsFromQueueTask).get();
+
+ // The client update message should still be delivered because it is now part of the
+ // filter clients interested in this event, despite having not been included in the original
+ // filter info in the "normal" put processing path.
+ verify(cacheClientProxy, times(1)).deliverMessage(clientUpdateMessage);
+ }
+
+ @Test
+ public void clientRemovedFromFilterClientsListIfEventAddedToRegistrationQueue() {
+ ClientRegistrationEventQueueManager clientRegistrationEventQueueManager =
+ new ClientRegistrationEventQueueManager();
+
+ ClientProxyMembershipID clientProxyMembershipID = mock(ClientProxyMembershipID.class);
+
+ clientRegistrationEventQueueManager.create(clientProxyMembershipID,
+ new ConcurrentLinkedQueue<>(), new ReentrantReadWriteLock());
+
+ InternalCacheEvent internalCacheEvent = mock(InternalCacheEvent.class);
+ when(internalCacheEvent.getRegion()).thenReturn(mock(LocalRegion.class));
+
+ Conflatable conflatable = mock(Conflatable.class);
+
+ // Add the registering client to the filter clients. This can happen if the filter info is
+ // received but the client is not completely registered yet (queue GII has not been completed).
+ // In that case, we want to remove the client from the filter IDs set and add the event
+ // to the client's registration queue.
+ Set<ClientProxyMembershipID> filterClientIDs = new HashSet<>();
+ filterClientIDs.add(clientProxyMembershipID);
+
+ CacheClientNotifier cacheClientNotifier = mock(CacheClientNotifier.class);
+
+ clientRegistrationEventQueueManager.add(internalCacheEvent, conflatable, filterClientIDs,
+ cacheClientNotifier);
+
+ // The client should no longer be in the filter clients since the event was queued in the
+ // client's registration queue.
+ assertThat(filterClientIDs.isEmpty()).isTrue();
+ }
+
+ @Test
+ public void addAndDrainQueueContentionTest() throws ExecutionException, InterruptedException {
+ ClientRegistrationEventQueueManager clientRegistrationEventQueueManager =
+ new ClientRegistrationEventQueueManager();
+
+ ClientProxyMembershipID clientProxyMembershipID = mock(ClientProxyMembershipID.class);
+ ReadWriteLock mockPutDrainLock = mock(ReadWriteLock.class);
+ ReadWriteLock actualPutDrainLock = new ReentrantReadWriteLock();
+
+ when(mockPutDrainLock.readLock())
+ .thenReturn(actualPutDrainLock.readLock());
+
+ when(mockPutDrainLock.writeLock())
+ .thenAnswer(i -> {
+ // Force a context switch from drain to put thread so we can ensure the event is not lost
+ Thread.sleep(1);
+ return actualPutDrainLock.writeLock();
+ });
+
+ ClientRegistrationEventQueueManager.ClientRegistrationEventQueue clientRegistrationEventQueue =
+ clientRegistrationEventQueueManager.create(clientProxyMembershipID,
+ new ConcurrentLinkedQueue<>(), mockPutDrainLock);
+
+ InternalCacheEvent internalCacheEvent = mock(InternalCacheEvent.class);
+ when(internalCacheEvent.getRegion()).thenReturn(mock(LocalRegion.class));
+
+ Conflatable conflatable = mock(Conflatable.class);
+ Set<ClientProxyMembershipID> filterClientIDs = new HashSet<>();
+ CacheClientNotifier cacheClientNotifier = mock(CacheClientNotifier.class);
+
+ CompletableFuture<Void> addEventsToQueueTask = CompletableFuture.runAsync(() -> {
+ for (int i = 0; i < 100000; ++i) {
+ // In thread one, we add events to the queue
+ clientRegistrationEventQueueManager
+ .add(internalCacheEvent, conflatable, filterClientIDs, cacheClientNotifier);
+ }
+ });
+
+ CompletableFuture<Void> drainEventsFromQueueTask = CompletableFuture.runAsync(() -> {
+ // In thread two, we drain events from the queue
+ clientRegistrationEventQueueManager.drain(clientProxyMembershipID, cacheClientNotifier);
+ });
+
+ CompletableFuture.allOf(addEventsToQueueTask, drainEventsFromQueueTask).get();
+
+ assertThat(clientRegistrationEventQueue.isEmpty()).isTrue();
+ }
+}
diff --git a/geode-junit/src/main/java/org/apache/geode/test/fake/Fakes.java b/geode-junit/src/main/java/org/apache/geode/test/fake/Fakes.java
index 23e534c..c1bd3f9 100644
--- a/geode-junit/src/main/java/org/apache/geode/test/fake/Fakes.java
+++ b/geode-junit/src/main/java/org/apache/geode/test/fake/Fakes.java
@@ -22,11 +22,11 @@ import static org.mockito.Mockito.when;
import java.io.File;
import java.util.HashMap;
import java.util.Map;
+import java.util.Properties;
import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
import org.apache.geode.CancelCriterion;
-import org.apache.geode.LogWriter;
import org.apache.geode.Statistics;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.DataPolicy;
@@ -38,9 +38,11 @@ import org.apache.geode.distributed.internal.DSClock;
import org.apache.geode.distributed.internal.DistributionConfig;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.SystemTimer;
import org.apache.geode.internal.cache.CachePerfStats;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.TXManagerImpl;
+import org.apache.geode.internal.logging.InternalLogWriter;
import org.apache.geode.internal.security.SecurableCommunicationChannel;
import org.apache.geode.internal.security.SecurityService;
import org.apache.geode.internal.statistics.StatisticsManager;
@@ -79,7 +81,7 @@ public class Fakes {
TypeRegistry pdxRegistryMock = mock(TypeRegistry.class);
CancelCriterion systemCancelCriterion = mock(CancelCriterion.class);
DSClock clock = mock(DSClock.class);
- LogWriter logger = mock(LogWriter.class);
+ InternalLogWriter logger = mock(InternalLogWriter.class);
Statistics stats = mock(Statistics.class);
TXManagerImpl txManager = mock(TXManagerImpl.class);
QueryMonitor queryMonitor = mock(QueryMonitor.class);
@@ -105,7 +107,7 @@ public class Fakes {
when(cache.getLogger()).thenReturn(logger);
when(cache.getQueryMonitor()).thenReturn(queryMonitor);
when(cache.getMeterRegistry()).thenReturn(new SimpleMeterRegistry());
-
+ when(cache.getCCPTimer()).thenReturn(mock(SystemTimer.class));
when(system.getDistributedMember()).thenReturn(member);
when(system.getConfig()).thenReturn(config);
when(system.getDistributionManager()).thenReturn(distributionManager);
@@ -117,6 +119,7 @@ public class Fakes {
when(system.getStatisticsManager()).thenReturn(statisticsManager);
when(system.createAtomicStatistics(any(), any(), anyLong())).thenReturn(stats);
when(system.createAtomicStatistics(any(), any())).thenReturn(stats);
+ when(system.getProperties()).thenReturn(mock(Properties.class));
when(distributionManager.getId()).thenReturn(member);
when(distributionManager.getDistributionManagerId()).thenReturn(member);