diff --git a/build.gradle b/build.gradle index 0d17209c99..f2d6e11d16 100644 --- a/build.gradle +++ b/build.gradle @@ -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' diff --git a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/HazelcastLocalInstanceRegistrar.java b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/HazelcastLocalInstanceRegistrar.java index d541c3f06d..877ea98e44 100644 --- a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/HazelcastLocalInstanceRegistrar.java +++ b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/HazelcastLocalInstanceRegistrar.java @@ -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()); } } diff --git a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/HazelcastIntegrationConfigurationInitializer.java b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/HazelcastIntegrationConfigurationInitializer.java deleted file mode 100644 index db0bfb6772..0000000000 --- a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/HazelcastIntegrationConfigurationInitializer.java +++ /dev/null @@ -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)); - } - } - -} diff --git a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/package-info.java b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/package-info.java deleted file mode 100644 index 450c9c0dde..0000000000 --- a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/package-info.java +++ /dev/null @@ -1,4 +0,0 @@ -/** - * Provides classes for configuration. - */ -package org.springframework.integration.hazelcast.config; diff --git a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/AbstractHazelcastMessageProducer.java b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/AbstractHazelcastMessageProducer.java index a3c295bf4e..01a91f81a2 100644 --- a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/AbstractHazelcastMessageProducer.java +++ b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/AbstractHazelcastMessageProducer.java @@ -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 hazelcastInstanceSet = Hazelcast.getAllHazelcastInstances(); - final Set localSocketAddressesSet = getLocalSocketAddresses(hazelcastInstanceSet); - return localSocketAddressesSet.isEmpty() || - localSocketAddressesSet.contains(socketAddress) - || isEventComingFromNonRegisteredHazelcastInstance(hazelcastInstanceSet.iterator().next(), - localSocketAddressesSet, socketAddress); - - } - - private Set getLocalSocketAddresses(final Set hazelcastInstanceSet) { - final Set localSocketAddressesSet = new HashSet<>(); + private boolean isEventLocal(InetSocketAddress socketAddress) { + Set 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 localSocketAddressesSet, - final InetSocketAddress socketAddressOfEvent) { - final MultiMap 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; } } diff --git a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/leader/LeaderInitiator.java b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/leader/LeaderInitiator.java index 726aee7035..de74bbe967 100644 --- a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/leader/LeaderInitiator.java +++ b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/leader/LeaderInitiator.java @@ -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); diff --git a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/listener/HazelcastMembershipListener.java b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/listener/HazelcastMembershipListener.java index 09da083a66..27f70e36b9 100644 --- a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/listener/HazelcastMembershipListener.java +++ b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/listener/HazelcastMembershipListener.java @@ -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 diff --git a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/lock/HazelcastLockRegistry.java b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/lock/HazelcastLockRegistry.java index bc5d9da696..0f1d70d4f8 100644 --- a/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/lock/HazelcastLockRegistry.java +++ b/spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/lock/HazelcastLockRegistry.java @@ -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; diff --git a/spring-integration-hazelcast/src/main/resources/META-INF/spring.factories b/spring-integration-hazelcast/src/main/resources/META-INF/spring.factories deleted file mode 100644 index d1db7a3bdb..0000000000 --- a/spring-integration-hazelcast/src/main/resources/META-INF/spring.factories +++ /dev/null @@ -1,2 +0,0 @@ -org.springframework.integration.config.IntegrationConfigurationInitializer=\ -org.springframework.integration.hazelcast.config.HazelcastIntegrationConfigurationInitializer diff --git a/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/common-config.xml b/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/common-config.xml index ff7256d52e..d841050b85 100644 --- a/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/common-config.xml +++ b/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/common-config.xml @@ -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"> - - - - - - - - - - - + diff --git a/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/inbound/config/HazelcastIntegrationInboundTestConfiguration.java b/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/inbound/config/HazelcastIntegrationInboundTestConfiguration.java index 15e8f5d10d..8c44a86a74 100644 --- a/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/inbound/config/HazelcastIntegrationInboundTestConfiguration.java +++ b/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/inbound/config/HazelcastIntegrationInboundTestConfiguration.java @@ -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 = diff --git a/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/leader/LeaderInitiatorTests.java b/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/leader/LeaderInitiatorTests.java deleted file mode 100644 index 514a8967dc..0000000000 --- a/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/leader/LeaderInitiatorTests.java +++ /dev/null @@ -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 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 { - - CountDownLatch onEventLatch = new CountDownLatch(1); - - ArrayList 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(); - } - - } - -} diff --git a/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/lock/HazelcastLockRegistryTests.java b/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/lock/HazelcastLockRegistryTests.java deleted file mode 100644 index 21346f39fd..0000000000 --- a/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/lock/HazelcastLockRegistryTests.java +++ /dev/null @@ -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 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 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(); - } - } - -} diff --git a/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/outbound/HazelcastOutboundChannelAdapterTests-context.xml b/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/outbound/HazelcastOutboundChannelAdapterTests-context.xml index c1638f0667..1c5d49bcee 100644 --- a/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/outbound/HazelcastOutboundChannelAdapterTests-context.xml +++ b/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/outbound/HazelcastOutboundChannelAdapterTests-context.xml @@ -221,8 +221,6 @@ - - @@ -271,10 +269,4 @@ - - - - - - diff --git a/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/outbound/HazelcastOutboundChannelAdapterTests.java b/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/outbound/HazelcastOutboundChannelAdapterTests.java index 7ad7bac317..cd40c4e49b 100644 --- a/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/outbound/HazelcastOutboundChannelAdapterTests.java +++ b/spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/outbound/HazelcastOutboundChannelAdapterTests.java @@ -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> 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++; diff --git a/src/reference/antora/modules/ROOT/pages/hazelcast.adoc b/src/reference/antora/modules/ROOT/pages/hazelcast.adoc index 2736d81c32..178562ee68 100644 --- a/src/reference/antora/modules/ROOT/pages/hazelcast.adoc +++ b/src/reference/antora/modules/ROOT/pages/hazelcast.adoc @@ -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`: diff --git a/src/reference/antora/modules/ROOT/pages/whats-new.adoc b/src/reference/antora/modules/ROOT/pages/whats-new.adoc index 3ca598cf34..a3cf9d71b7 100644 --- a/src/reference/antora/modules/ROOT/pages/whats-new.adoc +++ b/src/reference/antora/modules/ROOT/pages/whats-new.adoc @@ -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]]