RemoveAll extends BaseCommand { @Override public void cmdExecute(final Message clientMessage, final ServerConnection serverConnection, final SecurityService securityService, long startp) throws IOException, InterruptedException { long start = startp; Part regionNamePart = null, numberOfKeysPart = null, keyPart = null; String regionName = null; int numberOfKeys = 0; Object key = null; Part eventPart = null; boolean replyWithMetaData = false; VersionedObjectList response = null; StringBuilder errMessage = new StringBuilder(); CachedRegionHelper crHelper = serverConnection.getCachedRegionHelper(); CacheServerStats stats = serverConnection.getCacheServerStats(); serverConnection.setAsTrue(REQUIRES_RESPONSE); serverConnection.setAsTrue(REQUIRES_CHUNKED_RESPONSE); { long oldStart = start; start = DistributionStats.getStatTime(); stats.incReadRemoveAllRequestTime(start - oldStart); } try { regionNamePart = clientMessage.getPart(0); regionName = regionNamePart.getCachedString(); if (regionName == null) { String txt = "The input region name for the removeAll request is null"; logger.warn("{} : {}", new Object[] {serverConnection.getName(), txt}); errMessage.append(txt); writeChunkedErrorResponse(clientMessage, MessageType.PUT_DATA_ERROR, errMessage.toString(), serverConnection); serverConnection.setAsTrue(RESPONDED); return; } LocalRegion region = (LocalRegion) serverConnection.getCache().getRegion(regionName); if (region == null) { String reason = " was not found during removeAll request"; writeRegionDestroyedEx(clientMessage, regionName, reason, serverConnection); serverConnection.setAsTrue(RESPONDED); return; } eventPart = clientMessage.getPart(1); ByteBuffer eventIdPartsBuffer = ByteBuffer.wrap(eventPart.getSerializedForm()); long threadId = EventID.readEventIdPartsFromOptmizedByteArray(eventIdPartsBuffer); long sequenceId = EventID.readEventIdPartsFromOptmizedByteArray(eventIdPartsBuffer); EventID eventId = new EventID(serverConnection.getEventMemberIDByteArray(), threadId, sequenceId); Breadcrumbs.setEventId(eventId); int flags = clientMessage.getPart(2).getInt(); boolean clientIsEmpty = (flags & PutAllOp.FLAG_EMPTY) != 0; boolean clientHasCCEnabled = (flags & PutAllOp.FLAG_CONCURRENCY_CHECKS) != 0; Object callbackArg = clientMessage.getPart(3).getObject(); numberOfKeysPart = clientMessage.getPart(4); numberOfKeys = numberOfKeysPart.getInt(); if (logger.isDebugEnabled()) { StringBuilder buffer = new StringBuilder(); buffer.append(serverConnection.getName()).append(": Received removeAll request from ") .append(serverConnection.getSocketString()).append(" for region ").append(regionName) .append(callbackArg != null ? (" callbackArg " + callbackArg) : "").append(" with ") .append(numberOfKeys).append(" keys."); logger.debug(buffer); } ArrayList<Object> keys = new ArrayList<Object>(numberOfKeys); ArrayList<VersionTag> retryVersions = new ArrayList<VersionTag>(numberOfKeys); for (int i = 0; i < numberOfKeys; i++) { keyPart = clientMessage.getPart(5 + i); key = keyPart.getStringOrObject(); if (key == null) { String txt = "One of the input keys for the removeAll request is null"; logger.warn("{} : {}", new Object[] {serverConnection.getName(), txt}); errMessage.append(txt); writeChunkedErrorResponse(clientMessage, MessageType.PUT_DATA_ERROR, errMessage.toString(), serverConnection); serverConnection.setAsTrue(RESPONDED); return; } if (clientMessage.isRetry()) { EventID entryEventId = new EventID(eventId, i); if (region instanceof PartitionedRegion) { PartitionedRegion pr = (PartitionedRegion) region; int bucketId = pr.getKeyInfo(key).getBucketId(); long entryThreadId = ThreadIdentifier.createFakeThreadIDForBulkOp(bucketId, entryEventId.getThreadID()); entryEventId = new EventID(entryEventId.getMembershipID(), entryThreadId, entryEventId.getSequenceID()); } VersionTag tag = findVersionTagsForRetriedBulkOp(region, entryEventId); retryVersions.add(tag); } else { retryVersions.add(null); } keys.add(key); } if (clientMessage.getNumberOfParts() == (5 + numberOfKeys + 1)) { int timeout = clientMessage.getPart(5 + numberOfKeys).getInt(); serverConnection.setRequestSpecificTimeout(timeout); } securityService.authorize(Resource.DATA, Operation.WRITE, regionName); AuthorizeRequest authzRequest = serverConnection.getAuthzRequest(); if (authzRequest != null) { if (DynamicRegionFactory.regionIsDynamicRegionList(regionName)) { authzRequest.createRegionAuthorize(regionName); } else { RemoveAllOperationContext removeAllContext = authzRequest.removeAllAuthorize(regionName, keys, callbackArg); callbackArg = removeAllContext.getCallbackArg(); } } response = region.basicBridgeRemoveAll(keys, retryVersions, serverConnection.getProxyID(), eventId, callbackArg); if (!region.getConcurrencyChecksEnabled() || clientIsEmpty || !clientHasCCEnabled) { if (logger.isTraceEnabled()) { logger.trace( "setting removeAll response to null. region-cc-enabled={}; clientIsEmpty={}; client-cc-enabled={}", region.getConcurrencyChecksEnabled(), clientIsEmpty, clientHasCCEnabled); } response = null; } if (region instanceof PartitionedRegion) { PartitionedRegion pr = (PartitionedRegion) region; if (pr.getNetworkHopType() != PartitionedRegion.NETWORK_HOP_NONE) { writeReplyWithRefreshMetadata(clientMessage, response, serverConnection, pr, pr.getNetworkHopType()); pr.clearNetworkHopData(); replyWithMetaData = true; } } } catch (RegionDestroyedException rde) { writeChunkedException(clientMessage, rde, serverConnection); serverConnection.setAsTrue(RESPONDED); return; } catch (ResourceException re) { writeChunkedException(clientMessage, re, serverConnection); serverConnection.setAsTrue(RESPONDED); return; } catch (PutAllPartialResultException pre) { writeChunkedException(clientMessage, pre, serverConnection); serverConnection.setAsTrue(RESPONDED); return; } catch (Exception ce) { checkForInterrupt(serverConnection, ce); writeChunkedException(clientMessage, ce, serverConnection); serverConnection.setAsTrue(RESPONDED); logger.warn(String.format("%s: Unexpected Exception", serverConnection.getName()), ce); return; } finally { long oldStart = start; start = DistributionStats.getStatTime(); stats.incProcessRemoveAllTime(start - oldStart); } if (logger.isDebugEnabled()) { logger.debug("{}: Sending removeAll response back to {} for region {}{}", serverConnection.getName(), serverConnection.getSocketString(), regionName, (logger.isTraceEnabled() ? ": " + response : "")); } if (!replyWithMetaData) { writeReply(clientMessage, response, serverConnection); } serverConnection.setAsTrue(RESPONDED); stats.incWriteRemoveAllResponseTime(DistributionStats.getStatTime() - start); } protected  RemoveAll(); static Command getCommand(); @Override void cmdExecute(final Message clientMessage, final ServerConnection serverConnection,
      final SecurityService securityService, long startp);  }
@Test public void noSecurityShouldSucceed() throws Exception { when(this.securityService.isClientSecurityRequired()).thenReturn(false); this.removeAll.cmdExecute(this.message, this.serverConnection, this.securityService, 0); verify(this.chunkedResponseMessage).sendChunk(eq(this.serverConnection)); }
@Test public void integratedSecurityShouldSucceedIfAuthorized() throws Exception { when(this.securityService.isClientSecurityRequired()).thenReturn(true); when(this.securityService.isIntegratedSecurity()).thenReturn(true); this.removeAll.cmdExecute(this.message, this.serverConnection, this.securityService, 0); for (Object key : KEYS) { verify(this.securityService).authorize(Resource.DATA, Operation.WRITE, REGION_NAME); } verify(this.chunkedResponseMessage).sendChunk(eq(this.serverConnection)); }
@Test public void integratedSecurityShouldFailIfNotAuthorized() throws Exception { when(this.securityService.isClientSecurityRequired()).thenReturn(true); when(this.securityService.isIntegratedSecurity()).thenReturn(true); for (Object key : KEYS) { doThrow(new NotAuthorizedException("")).when(this.securityService).authorize(Resource.DATA, Operation.READ, REGION_NAME, key.toString()); } this.removeAll.cmdExecute(this.message, this.serverConnection, this.securityService, 0); for (Object key : KEYS) { verify(this.securityService).authorize(Resource.DATA, Operation.WRITE, REGION_NAME); } verify(this.chunkedResponseMessage).sendChunk(eq(this.serverConnection)); }
@Test public void oldSecurityShouldSucceedIfAuthorized() throws Exception { when(this.securityService.isClientSecurityRequired()).thenReturn(true); when(this.securityService.isIntegratedSecurity()).thenReturn(false); this.removeAll.cmdExecute(this.message, this.serverConnection, this.securityService, 0); for (Object key : KEYS) { verify(this.authzRequest).removeAllAuthorize(eq(REGION_NAME), any(), any()); } verify(this.chunkedResponseMessage).sendChunk(eq(this.serverConnection)); }
@Test public void oldSecurityShouldFailIfNotAuthorized() throws Exception { when(this.securityService.isClientSecurityRequired()).thenReturn(true); when(this.securityService.isIntegratedSecurity()).thenReturn(false); for (Object key : KEYS) { doThrow(new NotAuthorizedException("")).when(this.authzRequest).getAuthorize(eq(REGION_NAME), eq(key.toString()), eq(null)); } this.removeAll.cmdExecute(this.message, this.serverConnection, this.securityService, 0); for (Object key : KEYS) { verify(this.authzRequest).removeAllAuthorize(eq(REGION_NAME), any(), any()); } verify(this.chunkedResponseMessage).sendChunk(eq(this.serverConnection)); }