Spring/spring-data-geode
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix bug in SimpleGemfireRepository.deleteAll running in a client/server topology with a server-side PARTITION Region.

Browse Source 
Resolves gh-512.
This commit is contained in:
John Blum
2021-06-02 16:15:39 -07:00
parent 62d4ce6ebd
commit 73f4306f89
5 changed files with 353 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

66
spring-data-geode/src/main/java/org/springframework/data/gemfire/repository/support/SimpleGemfireRepository.java
View File

@@ -26,9 +26,9 @@ import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheTransactionManager;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.GemFireCache;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.query.SelectResults;
@@ -42,6 +42,7 @@ import org.springframework.data.gemfire.repository.Wrapper;
import org.springframework.data.gemfire.repository.query.QueryString;
import org.springframework.data.gemfire.repository.query.support.PagingUtils;
import org.springframework.data.gemfire.util.CollectionUtils;
import org.springframework.data.gemfire.util.RegionUtils;
import org.springframework.data.gemfire.util.SpringUtils;
import org.springframework.data.repository.CrudRepository;
import org.springframework.data.repository.PagingAndSortingRepository;
@@ -144,6 +145,9 @@ public class SimpleGemfireRepository<T, ID> implements GemfireRepository<T, ID>
return this.template;
}
/**
* @inheritDoc
*/
@Override
public <U extends T> U save(@NonNull U entity) {
@@ -159,6 +163,9 @@ public class SimpleGemfireRepository<T, ID> implements GemfireRepository<T, ID>
return entity;
}
/**
* @inheritDoc
*/
@Override
public T save(@NonNull Wrapper<T, ID> wrapper) {
@@ -174,6 +181,9 @@ public class SimpleGemfireRepository<T, ID> implements GemfireRepository<T, ID>
return entity;
}
/**
* @inheritDoc
*/
@Override
public <U extends T> Iterable<U> saveAll(@NonNull Iterable<U> entities) {
@@ -227,6 +237,9 @@ public class SimpleGemfireRepository<T, ID> implements GemfireRepository<T, ID>
return findById(id).isPresent();
}
/**
* @inheritDoc
*/
@Override
public @NonNull Iterable<T> findAll() {
@@ -238,6 +251,9 @@ public class SimpleGemfireRepository<T, ID> implements GemfireRepository<T, ID>
return toList(selectResults);
}
/**
* @inheritDoc
*/
@Override
public Page<T> findAll(@NonNull Pageable pageable) {
@@ -248,6 +264,9 @@ public class SimpleGemfireRepository<T, ID> implements GemfireRepository<T, ID>
return toPage(results, pageable);
}
/**
* @inheritDoc
*/
@Override
public @NonNull Iterable<T> findAll(@NonNull Sort sort) {
@@ -260,6 +279,9 @@ public class SimpleGemfireRepository<T, ID> implements GemfireRepository<T, ID>
return toList(selectResults);
}
/**
* @inheritDoc
*/
@Override
public @NonNull Iterable<T> findAllById(@NonNull Iterable<ID> ids) {
@@ -278,6 +300,9 @@ public class SimpleGemfireRepository<T, ID> implements GemfireRepository<T, ID>
return values;
}
/**
* @inheritDoc
*/
@Override
public Optional<T> findById(@NonNull ID id) {
@@ -288,11 +313,17 @@ public class SimpleGemfireRepository<T, ID> implements GemfireRepository<T, ID>
return Optional.ofNullable(value);
}
/**
* @inheritDoc
*/
@Override
public void delete(@NonNull T entity) {
deleteById(getEntityInformation().getRequiredId(entity));
}
/**
* @inheritDoc
*/
@Override
public void deleteAll() {
@@ -309,11 +340,17 @@ public class SimpleGemfireRepository<T, ID> implements GemfireRepository<T, ID>
});
}
/**
* @inheritDoc
*/
@Override
public void deleteAll(@NonNull Iterable<? extends T> entities) {
CollectionUtils.nullSafeIterable(entities).forEach(this::delete);
}
/**
* @inheritDoc
*/
@Override
public void deleteAllById(@NonNull Iterable<? extends ID> ids) {
@@ -327,34 +364,45 @@ public class SimpleGemfireRepository<T, ID> implements GemfireRepository<T, ID>
}
}
/**
* @inheritDoc
*/
@Override
public void deleteById(@NonNull ID id) {
getTemplate().remove(id);
}
boolean isPartitioned(Region<?, ?> region) {
boolean isPartitioned(@Nullable Region<?, ?> region) {
return region != null
&& region.getAttributes() != null
&& isPartitioned(region.getAttributes().getDataPolicy());
}
boolean isPartitioned(DataPolicy dataPolicy) {
boolean isPartitioned(@Nullable DataPolicy dataPolicy) {
return dataPolicy != null && dataPolicy.withPartitioning();
}
boolean isTransactionPresent(Region<?, ?> region) {
boolean isTransactionPresent(@Nullable Region<?, ?> region) {
return region.getRegionService() instanceof Cache
&& isTransactionPresent(((Cache) region.getRegionService()).getCacheTransactionManager());
return region != null
&& region.getRegionService() instanceof GemFireCache
&& isTransactionPresent(((GemFireCache) region.getRegionService()).getCacheTransactionManager());
}
boolean isTransactionPresent(CacheTransactionManager cacheTransactionManager) {
boolean isTransactionPresent(@Nullable CacheTransactionManager cacheTransactionManager) {
return cacheTransactionManager != null && cacheTransactionManager.exists();
}
<K> void doRegionClear(Region<K, ?> region) {
region.removeAll(region.keySet());
<K> void doRegionClear(@NonNull Region<K, ?> region) {
region.removeAll(resolveRegionKeys(region));
}
@NonNull <K> Set<K> resolveRegionKeys(@NonNull Region<K, ?> region) {
return RegionUtils.isClient(region) ? region.keySetOnServer()
: RegionUtils.isServer(region) ? region.keySet()
: Collections.emptySet();
}
@NonNull List<T> toList(@Nullable Iterable<T> iterable) {

34
spring-data-geode/src/main/java/org/springframework/data/gemfire/util/CacheUtils.java
View File

@@ -28,6 +28,8 @@ import org.apache.geode.cache.client.Pool;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.springframework.lang.NonNull;
import org.springframework.lang.Nullable;
import org.springframework.util.StringUtils;
/**
@@ -51,10 +53,9 @@ public abstract class CacheUtils extends DistributedSystemUtils {
public static final String DEFAULT_POOL_NAME = "DEFAULT";
@SuppressWarnings("all")
public static boolean isClient(GemFireCache cache) {
public static boolean isClient(@Nullable GemFireCache cache) {
boolean client = (cache instanceof ClientCache);
boolean client = cache instanceof ClientCache;
if (cache instanceof GemFireCacheImpl) {
client &= ((GemFireCacheImpl) cache).isClient();
@@ -63,23 +64,27 @@ public abstract class CacheUtils extends DistributedSystemUtils {
return client;
}
public static boolean isDefaultPool(Pool pool) {
return Optional.ofNullable(pool).map(Pool::getName).filter(CacheUtils::isDefaultPool).isPresent();
public static boolean isDefaultPool(@Nullable Pool pool) {
return Optional.ofNullable(pool)
.map(Pool::getName)
.filter(CacheUtils::isDefaultPool)
.isPresent();
}
public static boolean isNotDefaultPool(Pool pool) {
public static boolean isNotDefaultPool(@Nullable Pool pool) {
return !isDefaultPool(pool);
}
public static boolean isDefaultPool(String poolName) {
public static boolean isDefaultPool(@Nullable String poolName) {
return DEFAULT_POOL_NAME.equals(poolName);
}
public static boolean isNotDefaultPool(String poolName) {
public static boolean isNotDefaultPool(@Nullable String poolName) {
return !isDefaultPool(poolName);
}
public static boolean isDurable(ClientCache clientCache) {
public static boolean isDurable(@Nullable ClientCache clientCache) {
// NOTE: Technically, the following code snippet would be more useful/valuable but is not "testable"!
//((InternalDistributedSystem) distributedSystem).getConfig().getDurableClientId();
@@ -93,10 +98,9 @@ public abstract class CacheUtils extends DistributedSystemUtils {
.isPresent();
}
@SuppressWarnings("all")
public static boolean isPeer(GemFireCache cache) {
public static boolean isPeer(@Nullable GemFireCache cache) {
boolean peer = (cache instanceof Cache);
boolean peer = cache instanceof Cache;
if (cache instanceof GemFireCacheImpl) {
peer &= !((GemFireCacheImpl) cache).isClient();
@@ -109,17 +113,17 @@ public abstract class CacheUtils extends DistributedSystemUtils {
return close(resolveGemFireCache());
}
public static boolean close(GemFireCache gemfireCache) {
public static boolean close(@NonNull GemFireCache gemfireCache) {
return close(gemfireCache, () -> {});
}
public static boolean close(GemFireCache gemfireCache, Runnable shutdownHook) {
public static boolean close(@NonNull GemFireCache gemfireCache, @Nullable Runnable shutdownHook) {
try {
gemfireCache.close();
return true;
}
catch (Exception ignore) {
catch (Throwable ignore) {
return false;
}
finally {

13
spring-data-geode/src/main/java/org/springframework/data/gemfire/util/RegionUtils.java
View File

@@ -113,7 +113,7 @@ public abstract class RegionUtils extends CacheUtils {
* @return a boolean indicating whether the target {@link Region} is a {@literal client} {@link Region}.
* @see org.apache.geode.cache.Region
*/
public static boolean isClient(Region<?, ?> region) {
public static boolean isClient(@Nullable Region<?, ?> region) {
return Optional.ofNullable(region)
.map(Region::getAttributes)
@@ -175,4 +175,15 @@ public abstract class RegionUtils extends CacheUtils {
public static String toRegionPath(String regionName) {
return String.format("%1$s%2$s", Region.SEPARATOR, regionName);
}
/**
* Determines whether the target {@link Region} is a {@literal server-side} {@link Region}.
*
* @param region {@link Region} to evaluate.
* @return a boolean indicating whether the target {@link Region} is a {@literal server-side} {@link Region}.
* @see org.apache.geode.cache.Region
*/
public static boolean isServer(@Nullable Region<?, ?> region) {
return region != null && !isClient(region);
}
}

213
spring-data-geode/src/test/java/org/springframework/data/gemfire/repository/support/SimpleGemfireRepositoryRegionDeleteAllIntegrationTests.java Normal file
View File

@@ -0,0 +1,213 @@
/*
* Copyright 2020 the original author or authors.
*
* Licensed 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
*
* https://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.springframework.data.gemfire.repository.support;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.Arrays;
import java.util.Collections;
import javax.annotation.Resource;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.GemFireCache;
import org.apache.geode.cache.Region;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Profile;
import org.springframework.context.support.PropertySourcesPlaceholderConfigurer;
import org.springframework.data.gemfire.LocalRegionFactoryBean;
import org.springframework.data.gemfire.PartitionedRegionFactoryBean;
import org.springframework.data.gemfire.ReplicatedRegionFactoryBean;
import org.springframework.data.gemfire.config.annotation.CacheServerApplication;
import org.springframework.data.gemfire.config.annotation.ClientCacheApplication;
import org.springframework.data.gemfire.config.annotation.ClientCacheConfigurer;
import org.springframework.data.gemfire.config.annotation.EnableEntityDefinedRegions;
import org.springframework.data.gemfire.config.annotation.EnablePdx;
import org.springframework.data.gemfire.process.ProcessWrapper;
import org.springframework.data.gemfire.repository.config.EnableGemfireRepositories;
import org.springframework.data.gemfire.support.ConnectionEndpoint;
import org.springframework.data.gemfire.test.support.ClientServerIntegrationTestsSupport;
import org.springframework.data.repository.CrudRepository;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringRunner;
import example.app.model.User;
import example.app.repo.UserRepository;
/**
* Integration Tests asserting the correct function of the {@link CrudRepository#deleteAll()} method
* as implemented by the {@link SimpleGemfireRepository} class in an Apache Geode client/server topology.
*
* @author John Blum
* @see org.junit.Test
* @see org.apache.geode.cache.Region
* @see org.springframework.data.gemfire.config.annotation.CacheServerApplication
* @see org.springframework.data.gemfire.config.annotation.ClientCacheApplication
* @see org.springframework.data.gemfire.repository.config.EnableGemfireRepositories
* @see org.springframework.data.gemfire.repository.support.SimpleGemfireRepository
* @see org.springframework.data.repository.CrudRepository
* @see org.springframework.test.context.ContextConfiguration
* @see org.springframework.test.context.junit4.SpringRunner
* @link <a herf="https://github.com/spring-projects/spring-data-geode/issues/512">CrudRepository.deleteAll not working</a>
* @since 2.6.0
*/
@RunWith(SpringRunner.class)
@ContextConfiguration(classes = SimpleGemfireRepositoryRegionDeleteAllIntegrationTests.GeodeClientTestConfiguration.class)
@SuppressWarnings("unused")
public class SimpleGemfireRepositoryRegionDeleteAllIntegrationTests extends ClientServerIntegrationTestsSupport {
private static ProcessWrapper gemfireServer;
@BeforeClass
public static void startGeodeServer() throws Exception {
int availablePort = findAvailablePort();
gemfireServer = run(GeodeServerTestConfiguration.class, "-Dspring.profiles.active=partition",
String.format("-D%s=%d", GEMFIRE_CACHE_SERVER_PORT_PROPERTY, availablePort));
waitForServerToStart("localhost", availablePort);
System.setProperty(GEMFIRE_CACHE_SERVER_PORT_PROPERTY, String.valueOf(availablePort));
}
@AfterClass
public static void stopGemFireServer() {
System.clearProperty(GEMFIRE_CACHE_SERVER_PORT_PROPERTY);
stop(gemfireServer);
}
@Resource(name = "Users")
private Region<Integer, User> users;
@Autowired
private UserRepository userRepository;
@Before
public void assertRegionConfiguration() {
assertThat(this.users).isNotNull();
assertThat(this.users.getName()).isEqualTo("Users");
assertThat(this.users.getAttributes()).isNotNull();
assertThat(this.users.getAttributes().getDataPolicy()).isEqualTo(DataPolicy.EMPTY);
assertThat(this.users.getAttributes().getPoolName()).isNotEmpty();
assertThat(this.users.keySet()).isEmpty();
assertThat(this.userRepository).isNotNull();
}
@Before
public void storeUsers() {
User jonDoe = User.as("jonDoe").identifiedBy(1);
User janeDoe = User.as("janeDoe").identifiedBy(2);
this.userRepository.saveAll(Arrays.asList(jonDoe, janeDoe));
assertThat(this.users.keySet()).isEmpty();
assertThat(this.users.keySetOnServer()).containsExactlyInAnyOrder(jonDoe.getId(), janeDoe.getId());
}
@Test
public void deleteAllIsSuccessful() {
assertThat(this.userRepository.count()).isEqualTo(2);
this.userRepository.deleteAll();
assertThat(this.userRepository.count()).isZero();
assertThat(this.users.keySet()).isEmpty();
assertThat(this.users.keySetOnServer()).isEmpty();
}
@ClientCacheApplication
@EnablePdx
@EnableEntityDefinedRegions(basePackageClasses = User.class)
@EnableGemfireRepositories(basePackageClasses = UserRepository.class)
static class GeodeClientTestConfiguration {
@Bean
static PropertySourcesPlaceholderConfigurer propertySourcesPlaceholderConfigurer() {
return new PropertySourcesPlaceholderConfigurer();
}
@Bean
ClientCacheConfigurer clientCachePoolPortConfigurer(
@Value("${" + GEMFIRE_CACHE_SERVER_PORT_PROPERTY + ":40404}") int port) {
return (bean, clientCacheFactoryBean) -> clientCacheFactoryBean.setServers(
Collections.singletonList(new ConnectionEndpoint("localhost", port)));
}
}
@CacheServerApplication
static class GeodeServerTestConfiguration {
public static void main(String[] args) {
AnnotationConfigApplicationContext applicationContext =
new AnnotationConfigApplicationContext(GeodeServerTestConfiguration.class);
applicationContext.registerShutdownHook();
}
@Bean("Users")
@Profile("local")
public LocalRegionFactoryBean<Object, Object> localRegion(GemFireCache gemfireCache) {
LocalRegionFactoryBean<Object, Object> users = new LocalRegionFactoryBean<>();
users.setCache(gemfireCache);
users.setPersistent(false);
return users;
}
@Bean("Users")
@Profile("partition")
public PartitionedRegionFactoryBean<Object, Object> partitionRegion(GemFireCache gemfireCache) {
PartitionedRegionFactoryBean<Object, Object> users = new PartitionedRegionFactoryBean<>();
users.setCache(gemfireCache);
users.setPersistent(false);
return users;
}
@Bean("Users")
@Profile("replicate")
public ReplicatedRegionFactoryBean<Object, Object> replicateRegion(GemFireCache gemfireCache) {
ReplicatedRegionFactoryBean<Object, Object> users = new ReplicatedRegionFactoryBean<>();
users.setCache(gemfireCache);
users.setClose(false);
users.setPersistent(false);
return users;
}
}
}

60
spring-data-geode/src/test/java/org/springframework/data/gemfire/repository/support/SimpleGemfireRepositoryUnitTests.java
View File

@@ -25,6 +25,7 @@ import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -615,7 +616,7 @@ public class SimpleGemfireRepositoryUnitTests {
verifyNoMoreInteractions(mockRegion, mockSelectResults, mockTemplate);
}
// Page Numbers 0 based indexed
// Page Numbers are 0 based indexed
private void assertPage(Page page, int pageNumber, int pageSize, int total, Sort orderBy, User... content) {
int totalPages = (total / pageSize) + (total % pageSize > 0 ? 1 : 0);
@@ -909,6 +910,11 @@ public class SimpleGemfireRepositoryUnitTests {
verify(mockRegion, times(2)).getAttributes();
verify(mockRegion, times(2)).getRegionService();
verify(mockRegion, times(1)).clear();
verify(mockRegion, never()).keySet();
verify(mockRegion, never()).keySetOnServer();
verify(mockRegion, never()).removeAll(any(Collection.class));
verifyNoMoreInteractions(mockRegion);
}
@Test
@@ -920,7 +926,7 @@ public class SimpleGemfireRepositoryUnitTests {
Set<Long> keys = new HashSet<>(Arrays.asList(1L, 2L, 3L));
doThrow(new UnsupportedOperationException("Not Implemented!")).when(mockRegion).clear();
doThrow(new UnsupportedOperationException("TEST")).when(mockRegion).clear();
when(mockRegion.keySet()).thenReturn(keys);
SimpleGemfireRepository<Animal, Long> gemfireRepository =
@@ -929,9 +935,11 @@ public class SimpleGemfireRepositoryUnitTests {
gemfireRepository.deleteAll();
verify(mockCache, times(1)).getCacheTransactionManager();
verify(mockRegion, times(2)).getAttributes();
verify(mockRegion, times(4)).getAttributes();
verify(mockRegion, times(2)).getRegionService();
verify(mockRegion, times(1)).clear();
verify(mockRegion, times(1)).keySet();
verify(mockRegion, never()).keySetOnServer();
verify(mockRegion, times(1)).removeAll(eq(keys));
}
@@ -952,9 +960,11 @@ public class SimpleGemfireRepositoryUnitTests {
gemfireRepository.deleteAll();
verify(mockCache, times(0)).getCacheTransactionManager();
verify(mockRegion, times(2)).getAttributes();
verify(mockRegion, times(0)).getRegionService();
verify(mockRegion, times(0)).clear();
verify(mockRegion, times(4)).getAttributes();
verify(mockRegion, never()).getRegionService();
verify(mockRegion, never()).clear();
verify(mockRegion, times(1)).keySet();
verify(mockRegion, never()).keySetOnServer();
verify(mockRegion, times(1)).removeAll(eq(keys));
}
@@ -975,12 +985,45 @@ public class SimpleGemfireRepositoryUnitTests {
gemfireRepository.deleteAll();
verify(mockCache, times(1)).getCacheTransactionManager();
verify(mockRegion, times(2)).getAttributes();
verify(mockRegion, times(4)).getAttributes();
verify(mockRegion, times(2)).getRegionService();
verify(mockRegion, times(0)).clear();
verify(mockRegion, never()).clear();
verify(mockRegion, times(1)).keySet();
verify(mockRegion, never()).keySetOnServer();
verify(mockRegion, times(1)).removeAll(eq(keys));
}
@Test
public void deleteAllWithKeySetOnServerWhenClientRegion() {
Cache mockCache = mockCache("MockCache", false);
Region<Long, Animal> mockRegion = mockRegion("MockRegion", mockCache, DataPolicy.EMPTY);
RegionAttributes<Long, Animal> mockRegionAttributes = mockRegion.getAttributes();
Set<Long> keys = new HashSet<>(Arrays.asList(1L, 2L, 3L));
doThrow(new UnsupportedOperationException("TEST")).when(mockRegion).clear();
doReturn("TestPool").when(mockRegionAttributes).getPoolName();
doReturn(keys).when(mockRegion).keySetOnServer();
SimpleGemfireRepository<Animal, Long> gemfireRepository =
new SimpleGemfireRepository<>(newGemfireTemplate(mockRegion), mockEntityInformation());
gemfireRepository.deleteAll();
verify(mockCache, times(1)).getCacheTransactionManager();
verify(mockRegion, times(4)).getAttributes();
verify(mockRegion, times(2)).getRegionService();
verify(mockRegion, times(1)).clear();
verify(mockRegion, times(1)).keySetOnServer();
verify(mockRegion, never()).keySet();
verify(mockRegion, times(1)).removeAll(eq(keys));
verifyNoMoreInteractions(mockRegion);
}
@Test
public void deleteAllByIdWithKeys() {
@@ -1052,6 +1095,7 @@ public class SimpleGemfireRepositoryUnitTests {
}
@Test
@SuppressWarnings("all")
public void toListFromIterable() {
Set<User> userSet = CollectionUtils.asSet(User.newUser("Jon Doe"), User.newUser("Jane Doe"));