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

GH-9441: Deprecate Hazelcast CP Subsystem usage

Browse Source 
Fixes: https://github.com/spring-projects/spring-integration/issues/9441

* Upgrade to Hazelcast `5.5`
* Deprecate `LeaderInitiator`, `HazelcastLockRegistry` since they cannot be used in Open Source due to Hazelcast CP Subsystem migration under commercial support
* Deprecate `HazelcastMembershipListener` and `HazelcastLocalInstanceRegistrar` in favor of Hazelcast `Cluster` API.
* Mention changes in the docs
This commit is contained in:
Artem Bilan
2025-04-16 15:56:28 -04:00
parent 0f225d86c1
commit b705a966f3
17 changed files with 46 additions and 772 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -73,7 +73,7 @@ ext {
greenmailVersion = '2.1.3'
groovyVersion = '4.0.26'
hamcrestVersion = '3.0'
hazelcastVersion = '5.4.0'
hazelcastVersion = '5.5.0'
hibernateVersion = '6.6.13.Final'
hsqldbVersion = '2.7.4'
h2Version = '2.3.232'

15
spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/HazelcastLocalInstanceRegistrar.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2022 the original author or authors.
* Copyright 2015-2025 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.
@@ -27,7 +27,6 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.integration.hazelcast.listener.HazelcastMembershipListener;
/**
* This class creates an internal configuration {@link MultiMap} to cache Hazelcast instances' socket
@@ -39,7 +38,10 @@ import org.springframework.integration.hazelcast.listener.HazelcastMembershipLis
* @author Artem Bilan
*
* @since 6.0
*
* @deprecated in favor of {@link com.hazelcast.cluster.Cluster} API.
*/
@Deprecated(forRemoval = true, since = "6.5")
public class HazelcastLocalInstanceRegistrar implements SmartInitializingSingleton {
private static final Log logger = LogFactory.getLog(HazelcastLocalInstanceRegistrar.class);
@@ -77,12 +79,15 @@ public class HazelcastLocalInstanceRegistrar implements SmartInitializingSinglet
this.hazelcastInstance = hazelcastInstance;
}
@SuppressWarnings("removal")
@Override
public void afterSingletonsInstantiated() {
if (this.hazelcastInstance == null) {
if (!Hazelcast.getAllHazelcastInstances().isEmpty()) {
HazelcastInstance anyHazelcastInstance = Hazelcast.getAllHazelcastInstances().iterator().next();
anyHazelcastInstance.getCluster().addMembershipListener(new HazelcastMembershipListener());
anyHazelcastInstance.getCluster()
.addMembershipListener(
new org.springframework.integration.hazelcast.listener.HazelcastMembershipListener());
syncConfigurationMultiMap(anyHazelcastInstance);
}
else {
@@ -91,7 +96,9 @@ public class HazelcastLocalInstanceRegistrar implements SmartInitializingSinglet
}
else {
syncConfigurationMultiMap(this.hazelcastInstance);
this.hazelcastInstance.getCluster().addMembershipListener(new HazelcastMembershipListener());
this.hazelcastInstance.getCluster()
.addMembershipListener(
new org.springframework.integration.hazelcast.listener.HazelcastMembershipListener());
}
}

45
spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/HazelcastIntegrationConfigurationInitializer.java
View File

@@ -1,45 +0,0 @@
/*
* Copyright 2015-2022 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.integration.hazelcast.config;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.integration.config.IntegrationConfigurationInitializer;
import org.springframework.integration.hazelcast.HazelcastLocalInstanceRegistrar;
/**
* The Hazelcast Integration infrastructure {@code beanFactory} initializer.
*
* @author Eren Avsarogullari
* @author Artem Bilan
*
* @since 6.0
*/
public class HazelcastIntegrationConfigurationInitializer implements IntegrationConfigurationInitializer {
@Override
public void initialize(ConfigurableListableBeanFactory beanFactory) throws BeansException {
BeanDefinitionRegistry beanDefinitionRegistry = (BeanDefinitionRegistry) beanFactory;
if (!beanDefinitionRegistry.containsBeanDefinition(HazelcastLocalInstanceRegistrar.BEAN_NAME)) {
beanDefinitionRegistry.registerBeanDefinition(HazelcastLocalInstanceRegistrar.BEAN_NAME,
new RootBeanDefinition(HazelcastLocalInstanceRegistrar.class));
}
}
}

4
spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/package-info.java
View File

@@ -1,4 +0,0 @@
/**
* Provides classes for configuration.
*/
package org.springframework.integration.hazelcast.config;

40
spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/AbstractHazelcastMessageProducer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2024 the original author or authors.
* Copyright 2015-2025 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.
@@ -17,10 +17,8 @@
package org.springframework.integration.hazelcast.inbound;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
@@ -33,14 +31,12 @@ import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.instance.EndpointQualifier;
import com.hazelcast.map.AbstractIMapEvent;
import com.hazelcast.map.MapEvent;
import com.hazelcast.multimap.MultiMap;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.hazelcast.CacheEventType;
import org.springframework.integration.hazelcast.CacheListeningPolicyType;
import org.springframework.integration.hazelcast.HazelcastHeaders;
import org.springframework.integration.hazelcast.HazelcastIntegrationDefinitionValidator;
import org.springframework.integration.hazelcast.HazelcastLocalInstanceRegistrar;
import org.springframework.integration.hazelcast.message.EntryEventMessagePayload;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
@@ -112,40 +108,20 @@ public abstract class AbstractHazelcastMessageProducer extends MessageProducerSu
protected void sendMessage(E event, InetSocketAddress socketAddress,
CacheListeningPolicyType cacheListeningPolicyType) {
if (CacheListeningPolicyType.ALL == cacheListeningPolicyType || isEventAcceptable(socketAddress)) {
if (CacheListeningPolicyType.ALL == cacheListeningPolicyType || isEventLocal(socketAddress)) {
AbstractHazelcastMessageProducer.this.sendMessage(toMessage(event));
}
}
private boolean isEventAcceptable(final InetSocketAddress socketAddress) {
final Set<HazelcastInstance> hazelcastInstanceSet = Hazelcast.getAllHazelcastInstances();
final Set<SocketAddress> localSocketAddressesSet = getLocalSocketAddresses(hazelcastInstanceSet);
return localSocketAddressesSet.isEmpty() ||
localSocketAddressesSet.contains(socketAddress)
|| isEventComingFromNonRegisteredHazelcastInstance(hazelcastInstanceSet.iterator().next(),
localSocketAddressesSet, socketAddress);
}
private Set<SocketAddress> getLocalSocketAddresses(final Set<HazelcastInstance> hazelcastInstanceSet) {
final Set<SocketAddress> localSocketAddressesSet = new HashSet<>();
private boolean isEventLocal(InetSocketAddress socketAddress) {
Set<HazelcastInstance> hazelcastInstanceSet = Hazelcast.getAllHazelcastInstances();
for (HazelcastInstance hazelcastInstance : hazelcastInstanceSet) {
localSocketAddressesSet.add(hazelcastInstance.getLocalEndpoint().getSocketAddress());
if (socketAddress.equals(hazelcastInstance.getLocalEndpoint().getSocketAddress())) {
return true;
}
}
return localSocketAddressesSet;
}
private boolean isEventComingFromNonRegisteredHazelcastInstance(
final HazelcastInstance hazelcastInstance,
final Set<SocketAddress> localSocketAddressesSet,
final InetSocketAddress socketAddressOfEvent) {
final MultiMap<SocketAddress, SocketAddress> configMultiMap = hazelcastInstance
.getMultiMap(HazelcastLocalInstanceRegistrar.SPRING_INTEGRATION_INTERNAL_CLUSTER_MULTIMAP);
return configMultiMap.size() > 0
&& !configMultiMap.values().contains(socketAddressOfEvent)
&& localSocketAddressesSet.contains(configMultiMap.keySet().iterator().next());
return false;
}
}

6
spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/leader/LeaderInitiator.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2024 the original author or authors.
* Copyright 2015-2025 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.
@@ -57,7 +57,11 @@ import org.springframework.util.Assert;
* @author Robert Höglund
* @author Christian Tzolov
* @author Emil Palm
*
* @deprecated with no replacement since this class relies on the CP Subsystem
* which is not Open Source anymore since Hazelcast 5.5.
*/
@Deprecated(forRemoval = true, since = "6.5")
public class LeaderInitiator implements SmartLifecycle, DisposableBean, ApplicationEventPublisherAware {
private static final LogAccessor logger = new LogAccessor(LeaderInitiator.class);

6
spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/listener/HazelcastMembershipListener.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2022 the original author or authors.
* Copyright 2015-2025 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.
@@ -36,7 +36,11 @@ import org.springframework.integration.hazelcast.HazelcastLocalInstanceRegistrar
* @author Artem Bilan
*
* @since 6.0
*
* @deprecated in favor of {@link com.hazelcast.cluster.Cluster} API.
*/
@Deprecated(forRemoval = true, since = "6.5")
@SuppressWarnings("removal")
public class HazelcastMembershipListener extends MembershipAdapter {
@Override

6
spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/lock/HazelcastLockRegistry.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2022 the original author or authors.
* Copyright 2017-2025 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.
@@ -27,7 +27,11 @@ import org.springframework.util.Assert;
* A {@link LockRegistry} implementation Hazelcast distributed locks.
*
* @author Artem Bilan
*
* @deprecated with no replacement since this class relies on the CP Subsystem
* which is not Open Source anymore since Hazelcast 5.5.
*/
@Deprecated(forRemoval = true, since = "6.5")
public class HazelcastLockRegistry implements LockRegistry {
private final HazelcastInstance client;

2
spring-integration-hazelcast/src/main/resources/META-INF/spring.factories
View File

@@ -1,2 +0,0 @@
org.springframework.integration.config.IntegrationConfigurationInitializer=\
org.springframework.integration.hazelcast.config.HazelcastIntegrationConfigurationInitializer

12
spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/common-config.xml
View File

@@ -3,16 +3,6 @@
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd">
<bean id="instance" class="com.hazelcast.core.Hazelcast" factory-method="newHazelcastInstance" destroy-method="shutdown">
<constructor-arg>
<bean class="com.hazelcast.config.Config">
<property name="CPSubsystemConfig">
<bean class="com.hazelcast.config.cp.CPSubsystemConfig">
<property name="CPMemberCount" value="0"/>
</bean>
</property>
</bean>
</constructor-arg>
</bean>
<bean id="instance" class="com.hazelcast.core.Hazelcast" factory-method="newHazelcastInstance" destroy-method="shutdown"/>
</beans>

8
spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/inbound/config/HazelcastIntegrationInboundTestConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2022 the original author or authors.
* Copyright 2015-2025 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.
@@ -35,7 +35,6 @@ import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.hazelcast.DistributedSQLIterationType;
import org.springframework.integration.hazelcast.HazelcastIntegrationTestUser;
import org.springframework.integration.hazelcast.HazelcastLocalInstanceRegistrar;
import org.springframework.integration.hazelcast.inbound.HazelcastClusterMonitorMessageProducer;
import org.springframework.integration.hazelcast.inbound.HazelcastContinuousQueryMessageProducer;
import org.springframework.integration.hazelcast.inbound.HazelcastDistributedSQLMessageSource;
@@ -225,11 +224,6 @@ public class HazelcastIntegrationInboundTestConfiguration {
return Hazelcast.newHazelcastInstance(hazelcastConfig());
}
@Bean(HazelcastLocalInstanceRegistrar.BEAN_NAME)
public HazelcastLocalInstanceRegistrar hazelcastLocalInstanceRegistrar() {
return new HazelcastLocalInstanceRegistrar(testHazelcastInstance());
}
@Bean
public HazelcastEventDrivenMessageProducer hazelcastEventDrivenMessageProducer() {
final HazelcastEventDrivenMessageProducer producer =

356
spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/leader/LeaderInitiatorTests.java
View File

@@ -1,356 +0,0 @@
/*
* Copyright 2015-2024 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.integration.hazelcast.leader;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import com.hazelcast.config.Config;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.cp.CPGroupId;
import com.hazelcast.cp.CPSubsystem;
import com.hazelcast.cp.lock.FencedLock;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.leader.Candidate;
import org.springframework.integration.leader.Context;
import org.springframework.integration.leader.DefaultCandidate;
import org.springframework.integration.leader.event.AbstractLeaderEvent;
import org.springframework.integration.leader.event.DefaultLeaderEventPublisher;
import org.springframework.integration.leader.event.LeaderEventPublisher;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.willAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
/**
* Tests for hazelcast leader election.
*
* @author Janne Valkealahti
* @author Patrick Peralta
* @author Dave Syer
* @author Artem Bilan
* @author Mael Le Guével
* @author Emil Palm
*/
@SpringJUnitConfig
@DirtiesContext
public class LeaderInitiatorTests {
@Autowired
private HazelcastInstance hazelcastInstance;
@Autowired
private TestCandidate candidate;
@Autowired
private TestEventListener listener;
@Autowired
private LeaderInitiator initiator;
@Test
public void testLeaderElections() throws Exception {
assertThat(this.candidate.onGrantedLatch.await(5, TimeUnit.SECONDS)).isTrue();
assertThat(this.listener.onEventLatch.await(5, TimeUnit.SECONDS)).isTrue();
assertThat(this.listener.events.size()).isEqualTo(1);
this.initiator.destroy();
CountDownLatch granted = new CountDownLatch(1);
CountingPublisher countingPublisher = new CountingPublisher(granted);
List<LeaderInitiator> initiators = new ArrayList<>();
for (int i = 0; i < 2; i++) {
LeaderInitiator initiator = new LeaderInitiator(this.hazelcastInstance, new DefaultCandidate());
initiator.setLeaderEventPublisher(countingPublisher);
initiators.add(initiator);
}
for (LeaderInitiator initiator : initiators) {
initiator.start();
}
assertThat(granted.await(10, TimeUnit.SECONDS)).isTrue();
LeaderInitiator initiator1 = countingPublisher.initiator;
LeaderInitiator initiator2 = null;
for (LeaderInitiator initiator : initiators) {
if (initiator != initiator1) {
initiator2 = initiator;
break;
}
}
assertThat(initiator2).isNotNull();
assertThat(initiator1.getContext().isLeader()).isTrue();
assertThat(initiator2.getContext().isLeader()).isFalse();
final CountDownLatch granted1 = new CountDownLatch(1);
final CountDownLatch granted2 = new CountDownLatch(1);
CountDownLatch revoked1 = new CountDownLatch(1);
CountDownLatch revoked2 = new CountDownLatch(1);
initiator1.setLeaderEventPublisher(new CountingPublisher(granted1, revoked1) {
@Override
public void publishOnRevoked(Object source, Context context, String role) {
try {
// It's difficult to see round-robin election, so block one initiator until the second is elected.
assertThat(granted2.await(10, TimeUnit.SECONDS)).isTrue();
}
catch (InterruptedException e) {
// No op
}
super.publishOnRevoked(source, context, role);
}
});
initiator2.setLeaderEventPublisher(new CountingPublisher(granted2, revoked2) {
@Override
public void publishOnRevoked(Object source, Context context, String role) {
try {
// It's difficult to see round-robin election, so block one initiator until the second is elected.
assertThat(granted1.await(10, TimeUnit.SECONDS)).isTrue();
}
catch (InterruptedException e) {
// No op
}
super.publishOnRevoked(source, context, role);
}
});
initiator1.getContext().yield();
assertThat(revoked1.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(initiator2.getContext().isLeader()).isTrue();
assertThat(initiator1.getContext().isLeader()).isFalse();
initiator2.getContext().yield();
assertThat(revoked2.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(initiator1.getContext().isLeader()).isTrue();
assertThat(initiator2.getContext().isLeader()).isFalse();
initiator2.destroy();
CountDownLatch revoked11 = new CountDownLatch(1);
initiator1.setLeaderEventPublisher(new CountingPublisher(new CountDownLatch(1), revoked11));
initiator1.getContext().yield();
assertThat(revoked11.await(10, TimeUnit.SECONDS)).isTrue();
initiator1.destroy();
CountDownLatch onGranted = new CountDownLatch(1);
DefaultCandidate candidate = spy(new DefaultCandidate());
willAnswer(invocation -> {
try {
return invocation.callRealMethod();
}
finally {
onGranted.countDown();
}
})
.given(candidate).onGranted(any(Context.class));
LeaderInitiator initiator = new LeaderInitiator(this.hazelcastInstance, candidate);
initiator.setLeaderEventPublisher(new DefaultLeaderEventPublisher() {
@Override
public void publishOnGranted(Object source, Context context, String role) {
throw new RuntimeException("intentional");
}
});
initiator.start();
assertThat(onGranted.await(5, TimeUnit.SECONDS)).isTrue();
assertThat(initiator.getContext().isLeader()).isTrue();
initiator.destroy();
}
@Test
public void testRevokeLeadershipCalledWhenLockNotAcquiredButStillLeader() throws Exception {
// Initialize mocks and objects needed for the revoke leadership when fenced lock is no longer acquired
HazelcastInstance hazelcastInstance = mock();
Candidate candidate = mock();
FencedLock fencedLock = mock();
LeaderEventPublisher leaderEventPublisher = mock();
CPSubsystem cpSubsystem = mock(CPSubsystem.class);
given(candidate.getRole()).willReturn("role");
given(hazelcastInstance.getCPSubsystem()).willReturn(cpSubsystem);
given(cpSubsystem.getLock(anyString())).willReturn(fencedLock);
given(fencedLock.getGroupId())
.willReturn(new CPGroupId() {
@Override
public String getName() {
return "";
}
@Override
public long getId() {
return 0;
}
});
LeaderInitiator leaderInitiator = new LeaderInitiator(hazelcastInstance, candidate);
leaderInitiator.setLeaderEventPublisher(leaderEventPublisher);
// Simulate that the lock is currently held by this thread
given(fencedLock.isLockedByCurrentThread()).willReturn(true, false);
given(fencedLock.tryLock(anyLong(), any(TimeUnit.class))).willReturn(false); // Lock acquisition fails
// Start the LeaderInitiator to trigger the leader election process
leaderInitiator.start();
// Simulate the lock acquisition check process
Thread.sleep(1000); // Give time for the async task to run
// Verify that revokeLeadership was called due to lock not being acquired
// unlock is part of revokeLeadership
verify(fencedLock).unlock();
// verify revoke event is published
verify(leaderEventPublisher).publishOnRevoked(any(Object.class), any(Context.class), anyString());
leaderInitiator.destroy();
}
@Configuration
public static class TestConfig {
@Bean
TestCandidate candidate() {
return new TestCandidate();
}
@Bean
Config hazelcastConfig() {
Config config = new Config();
config.getCPSubsystemConfig()
.setSessionHeartbeatIntervalSeconds(1);
return config;
}
@Bean(destroyMethod = "shutdown")
HazelcastInstance hazelcastInstance() {
return Hazelcast.newHazelcastInstance(hazelcastConfig());
}
@Bean
LeaderInitiator initiator() {
return new LeaderInitiator(hazelcastInstance(), candidate());
}
@Bean
TestEventListener testEventListener() {
return new TestEventListener();
}
}
static class TestCandidate extends DefaultCandidate {
CountDownLatch onGrantedLatch = new CountDownLatch(1);
@Override
public void onGranted(Context ctx) {
this.onGrantedLatch.countDown();
super.onGranted(ctx);
}
}
static class TestEventListener implements ApplicationListener<AbstractLeaderEvent> {
CountDownLatch onEventLatch = new CountDownLatch(1);
ArrayList<AbstractLeaderEvent> events = new ArrayList<>();
@Override
public void onApplicationEvent(AbstractLeaderEvent event) {
this.events.add(event);
this.onEventLatch.countDown();
}
}
private static class CountingPublisher implements LeaderEventPublisher {
private CountDownLatch granted;
private CountDownLatch revoked;
private volatile LeaderInitiator initiator;
CountingPublisher(CountDownLatch granted, CountDownLatch revoked) {
this.granted = granted;
this.revoked = revoked;
}
CountingPublisher(CountDownLatch granted) {
this(granted, new CountDownLatch(1));
}
@Override
public void publishOnRevoked(Object source, Context context, String role) {
this.revoked.countDown();
}
@Override
public void publishOnFailedToAcquire(Object source, Context context, String role) {
}
@Override
public void publishOnGranted(Object source, Context context, String role) {
this.initiator = (LeaderInitiator) source;
this.granted.countDown();
}
}
}

284
spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/lock/HazelcastLockRegistryTests.java
View File

@@ -1,284 +0,0 @@
/*
* Copyright 2017-2024 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.integration.hazelcast.lock;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import com.hazelcast.config.Config;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.cp.lock.FencedLock;
import com.hazelcast.instance.impl.HazelcastInstanceFactory;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Artem Bilan
*/
public class HazelcastLockRegistryTests {
private static final Config CONFIG = new Config();
static {
CONFIG.getCPSubsystemConfig().setCPMemberCount(0);
}
private static final HazelcastInstance instance = Hazelcast.newHazelcastInstance(CONFIG);
@AfterAll
public static void destroy() {
HazelcastInstanceFactory.terminateAll();
}
@Test
public void testLock() {
HazelcastLockRegistry registry = new HazelcastLockRegistry(instance);
for (int i = 0; i < 10; i++) {
Lock lock = registry.obtain("foo");
lock.lock();
try {
assertThat(((FencedLock) lock).isLocked()).isTrue();
assertThat(((FencedLock) lock).isLockedByCurrentThread()).isTrue();
}
finally {
lock.unlock();
}
}
}
@Test
public void testLockInterruptibly() throws Exception {
HazelcastLockRegistry registry = new HazelcastLockRegistry(instance);
for (int i = 0; i < 10; i++) {
Lock lock = registry.obtain("foo");
lock.lockInterruptibly();
try {
assertThat(((FencedLock) lock).isLocked()).isTrue();
assertThat(((FencedLock) lock).isLockedByCurrentThread()).isTrue();
}
finally {
lock.unlock();
}
}
}
@Test
public void testReentrantLock() {
HazelcastLockRegistry registry = new HazelcastLockRegistry(instance);
for (int i = 0; i < 10; i++) {
Lock lock1 = registry.obtain("foo");
lock1.lock();
try {
Lock lock2 = registry.obtain("foo");
assertThat(lock2).isSameAs(lock1);
lock2.lock();
lock2.unlock();
}
finally {
lock1.unlock();
}
}
}
@Test
public void testReentrantLockInterruptibly() throws Exception {
HazelcastLockRegistry registry = new HazelcastLockRegistry(instance);
for (int i = 0; i < 10; i++) {
Lock lock1 = registry.obtain("foo");
lock1.lockInterruptibly();
try {
Lock lock2 = registry.obtain("foo");
assertThat(lock2).isSameAs(lock1);
lock2.lockInterruptibly();
lock2.unlock();
}
finally {
lock1.unlock();
}
}
}
@Test
public void testTwoLocks() throws Exception {
HazelcastLockRegistry registry = new HazelcastLockRegistry(instance);
for (int i = 0; i < 10; i++) {
Lock lock1 = registry.obtain("foo");
lock1.lockInterruptibly();
try {
Lock lock2 = registry.obtain("bar");
assertThat(lock2).isNotSameAs(lock1);
lock2.lockInterruptibly();
lock2.unlock();
}
finally {
lock1.unlock();
}
}
}
@Test
public void testTwoThreadsSecondFailsToGetLock() throws Exception {
HazelcastLockRegistry registry = new HazelcastLockRegistry(instance);
Lock lock1 = registry.obtain("foo");
lock1.lockInterruptibly();
AtomicBoolean locked = new AtomicBoolean();
CountDownLatch latch = new CountDownLatch(1);
ExecutorService executorService = Executors.newSingleThreadExecutor();
Future<Object> result = executorService.submit(() -> {
Lock lock2 = registry.obtain("foo");
locked.set(lock2.tryLock(200, TimeUnit.MILLISECONDS));
latch.countDown();
try {
lock2.unlock();
}
catch (Exception e) {
return e;
}
return null;
});
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(locked.get()).isFalse();
lock1.unlock();
Object ise = result.get(10, TimeUnit.SECONDS);
assertThat(ise).isInstanceOf(IllegalMonitorStateException.class);
assertThat(((Exception) ise).getMessage()).contains("Current thread is not owner of the lock!");
executorService.shutdown();
}
@Test
public void testTwoThreads() throws Exception {
HazelcastLockRegistry registry = new HazelcastLockRegistry(instance);
Lock lock1 = registry.obtain("foo");
AtomicBoolean locked = new AtomicBoolean();
CountDownLatch latch1 = new CountDownLatch(1);
CountDownLatch latch2 = new CountDownLatch(1);
CountDownLatch latch3 = new CountDownLatch(1);
lock1.lockInterruptibly();
ExecutorService executorService = Executors.newSingleThreadExecutor();
executorService.execute(() -> {
Lock lock2 = registry.obtain("foo");
try {
latch1.countDown();
lock2.lockInterruptibly();
latch2.await(10, TimeUnit.SECONDS);
locked.set(true);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
finally {
lock2.unlock();
latch3.countDown();
}
});
assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(locked.get()).isFalse();
lock1.unlock();
latch2.countDown();
assertThat(latch3.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(locked.get()).isTrue();
executorService.shutdown();
}
@Test
public void testTwoThreadsDifferentRegistries() throws Exception {
HazelcastLockRegistry registry1 = new HazelcastLockRegistry(instance);
HazelcastLockRegistry registry2 = new HazelcastLockRegistry(instance);
Lock lock1 = registry1.obtain("foo");
AtomicBoolean locked = new AtomicBoolean();
CountDownLatch latch1 = new CountDownLatch(1);
CountDownLatch latch2 = new CountDownLatch(1);
CountDownLatch latch3 = new CountDownLatch(1);
lock1.lockInterruptibly();
ExecutorService executorService = Executors.newSingleThreadExecutor();
executorService.execute(() -> {
Lock lock2 = registry2.obtain("foo");
try {
latch1.countDown();
lock2.lockInterruptibly();
latch2.await(10, TimeUnit.SECONDS);
locked.set(true);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
finally {
lock2.unlock();
latch3.countDown();
}
});
assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(locked.get()).isFalse();
lock1.unlock();
latch2.countDown();
assertThat(latch3.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(locked.get()).isTrue();
executorService.shutdown();
}
@Test
public void testTwoThreadsWrongOneUnlocks() throws Exception {
HazelcastLockRegistry registry = new HazelcastLockRegistry(instance);
final Lock lock = registry.obtain("foo");
lock.lockInterruptibly();
final AtomicBoolean locked = new AtomicBoolean();
final CountDownLatch latch = new CountDownLatch(1);
ExecutorService executorService = Executors.newSingleThreadExecutor();
Future<Object> result = executorService.submit(() -> {
try {
lock.unlock();
}
catch (Exception e) {
latch.countDown();
return e;
}
return null;
});
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(locked.get()).isFalse();
lock.unlock();
Object imse = result.get(10, TimeUnit.SECONDS);
assertThat(imse).isInstanceOf(IllegalMonitorStateException.class);
assertThat(((Exception) imse).getMessage()).contains("Current thread is not owner of the lock!");
executorService.shutdown();
}
@Test
public void testTryLock() throws Exception {
HazelcastLockRegistry registry = new HazelcastLockRegistry(instance);
for (int i = 0; i < 10; i++) {
Lock lock = registry.obtain("foo");
int n = 0;
while (!lock.tryLock() && n++ < 100) {
Thread.sleep(100);
}
assertThat(n).isLessThan(100);
lock.unlock();
}
}
}

8
spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/outbound/HazelcastOutboundChannelAdapterTests-context.xml
View File

@@ -221,8 +221,6 @@
</int-hazelcast:request-handler-advice-chain>
</int-hazelcast:outbound-channel-adapter>
<int-hazelcast:outbound-channel-adapter id="lockChannel" cache="myLock" />
<bean id="distributedMap" factory-bean="instance" factory-method="getMap">
<constructor-arg value="distributedMap"/>
</bean>
@@ -271,10 +269,4 @@
<constructor-arg value="topic"/>
</bean>
<bean id="cpSubsystem" factory-bean="instance" factory-method="getCPSubsystem"/>
<bean id="myLock" factory-bean="cpSubsystem" factory-method="getLock">
<constructor-arg value="myLock"/>
</bean>
</beans>

14
spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/outbound/HazelcastOutboundChannelAdapterTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2024 the original author or authors.
* Copyright 2015-2025 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.
@@ -143,10 +143,6 @@ public class HazelcastOutboundChannelAdapterTests {
@Qualifier("topicChannel")
private MessageChannel topicChannel;
@Autowired
@Qualifier("lockChannel")
private MessageChannel lockChannel;
@Autowired
private Map<?, ?> distributedMap;
@@ -437,12 +433,6 @@ public class HazelcastOutboundChannelAdapterTests {
.isThrownBy(() -> this.sixthMapChannel.send(message));
}
@Test
public void testWriteToLock() {
assertThatExceptionOfType(MessageHandlingException.class)
.isThrownBy(() -> this.lockChannel.send(new GenericMessage<>("foo")));
}
private void sendMessageWithCacheHeaderToChannel(final MessageChannel channel,
final String headerName, final String distributedObjectName) {
@@ -461,7 +451,7 @@ public class HazelcastOutboundChannelAdapterTests {
for (Entry<Integer, Message<HazelcastIntegrationTestUser>> entry : map.entrySet()) {
assertThat(entry).isNotNull();
assertThat(entry.getKey().intValue()).isEqualTo(index);
assertThat(entry.getValue().getHeaders().size() > 0).isTrue();
assertThat(!entry.getValue().getHeaders().isEmpty()).isTrue();
HazelcastOutboundChannelAdapterTestUtils
.verifyHazelcastIntegrationTestUser(entry.getValue().getPayload(), index);
index++;

6
src/reference/antora/modules/ROOT/pages/hazelcast.adoc
View File

@@ -528,6 +528,8 @@ public LeaderInitiator initiator() {
When a node is elected leader it will send an `OnGrantedEvent` to all application listeners.
NOTE: The `LeaderInitiator` has been deprecated since version 6.5, since it relies on CP Subsystem which had been moved out of Open Source since Hazelcast 5.5.
[[hazelcast-message-store]]
== Hazelcast Message Store
@@ -589,6 +591,8 @@ public LockRegistry lockRegistry() {
}
----
NOTE: The `HazelcastLockRegistry` has been deprecated since version 6.5, since it relies on CP Subsystem which had been moved out of Open Source since Hazelcast 5.5.
When used with a shared `MessageGroupStore` (e.g. `Aggregator` store management), the `HazelcastLockRegistry` can be used to provide this functionality across multiple application instances, such that only one instance can manipulate the group at a time.
NOTE: For all the distributed operations the CP Subsystem must be enabled on `HazelcastInstance`.
@@ -596,7 +600,7 @@ NOTE: For all the distributed operations the CP Subsystem must be enabled on `Ha
[[hazelcast-message-channels]]
== Message Channels with Hazelcast
The Hazelcast `IQueue` and `ITopic` distributed objects are, essentially, messaging primitives and can be use with Spring Integration core components without extra implementations in this Hazelcast module.
The Hazelcast `IQueue` and `ITopic` distributed objects are, essentially, messaging primitives and can be used with Spring Integration core components without extra implementations in this Hazelcast module.
The xref:channel/implementations.adoc#channel-implementations-queuechannel[`QueueChannel`] can be supplied by any `java.util.Queue`, including the mentioned Hazelcast distributed `IQueue`:

4
src/reference/antora/modules/ROOT/pages/whats-new.adoc
View File

@@ -77,9 +77,9 @@ The respective implementations are provided: `RecentFileListFilter`, `FtpRecentF
See xref:file/reading.adoc[Reading Files] for more information.
[[x6.5-hazelcast-changes]]
== Hazelcast Support Deprecation
== Hazelcast Module Deprecations
The `spring-integration-hazelcast` module has been deprecated due to Hazelcast migration to Enterprise Edition.
The `HazelcastLockRegistry` and Hazelcast `LeaderInitiator` have been deprecated due to Hazelcast CP Subsystem migration to Enterprise Edition.
See xref:hazelcast.adoc[Hazelcast Support] for more information.
[[x6.5-jdbc-changes]]