Move spring-integration-hazelcast from extensions

* Update javadocs
* Upgrade from log4j to log4j2
* Remove unnecessary test deps.
* Change from implmementation to api
* Remove unneeded test deps.
* Fix HZ XSD to be "versionless"
* Use `com.google.code.findbugs:annotations` dep to avoid warning from HZ code base

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/CacheEventType.java

@@ -0,0 +1,66 @@
+/*
+ * 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;
+
+/**
+ * Enumeration of Cache Event Types.
+ *
+ * @author Eren Avsarogullari
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ *
+ * @see org.springframework.integration.hazelcast.inbound.AbstractHazelcastMessageProducer
+ */
+public enum CacheEventType {
+
+	/**
+	 * The Hazelcast ADDED event.
+	 */
+	ADDED,
+
+	/**
+	 * The Hazelcast REMOVED event.
+	 */
+	REMOVED,
+
+	/**
+	 * The Hazelcast UPDATED event.
+	 */
+	UPDATED,
+
+	/**
+	 * The Hazelcast EVICTED event.
+	 */
+	EVICTED,
+
+	/**
+	 * The Hazelcast EXPIRED event.
+	 */
+	EXPIRED,
+
+	/**
+	 * The Hazelcast EVICT_ALL event.
+	 */
+	EVICT_ALL,
+
+	/**
+	 * The Hazelcast CLEAR_ALL event.
+	 */
+	CLEAR_ALL
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/CacheListeningPolicyType.java

@@ -0,0 +1,41 @@
+/*
+ * 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;
+
+/**
+ * Enumeration of Cache Listening Policy Type.
+ *
+ * @author Eren Avsarogullari
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ *
+ * @see org.springframework.integration.hazelcast.inbound.AbstractHazelcastMessageProducer
+ */
+public enum CacheListeningPolicyType {
+
+	/**
+	 * Only the local Hazelcast node can accept event.
+	 */
+	SINGLE,
+
+	/**
+	 * All subscribed members can accept event.
+	 */
+	ALL
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/ClusterMonitorType.java

@@ -0,0 +1,58 @@
+/*
+ * 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;
+
+/**
+ * Enumeration of Hazelcast Cluster Monitor Types.
+ *
+ * @author Eren Avsarogullari
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ *
+ * @see org.springframework.integration.hazelcast.inbound.HazelcastClusterMonitorMessageProducer
+ * @see com.hazelcast.core.DistributedObjectListener
+ * @see com.hazelcast.core.LifecycleListener
+ */
+public enum ClusterMonitorType {
+
+	/**
+	 * The membership listener mode.
+	 */
+	MEMBERSHIP,
+
+	/**
+	 * The distributed object listener mode.
+	 */
+	DISTRIBUTED_OBJECT,
+
+	/**
+	 * The migration listener mode.
+	 */
+	MIGRATION,
+
+	/**
+	 * The listener listener mode.
+	 */
+	LIFECYCLE,
+
+	/**
+	 * The client listener mode.
+	 */
+	CLIENT
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/DistributedSQLIterationType.java

@@ -0,0 +1,52 @@
+/*
+ * 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;
+
+/**
+ * Enumeration of Distributed SQL Iteration Type.
+ *
+ * @author Eren Avsarogullari
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ *
+ * @see org.springframework.integration.hazelcast.inbound.HazelcastDistributedSQLMessageSource
+ * @see com.hazelcast.map.IMap
+ */
+public enum DistributedSQLIterationType {
+
+	/**
+	 * The {@link com.hazelcast.map.IMap#entrySet()} to iterate.
+	 */
+	ENTRY,
+
+	/**
+	 * The {@link com.hazelcast.map.IMap#keySet()} to iterate.
+	 */
+	KEY,
+
+	/**
+	 * The {@link com.hazelcast.map.IMap#localKeySet()} to iterate.
+	 */
+	LOCAL_KEY,
+
+	/**
+	 * The {@link com.hazelcast.map.IMap#values()} to iterate.
+	 */
+	VALUE
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/HazelcastHeaders.java

@@ -0,0 +1,51 @@
+/*
+ * 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;
+
+/**
+ * Hazelcast Message Headers.
+ *
+ * @author Eren Avsarogullari
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+public abstract class HazelcastHeaders {
+
+	private static final String PREFIX = "hazelcast_";
+
+	/**
+	 * The event type header name.
+	 */
+	public static final String EVENT_TYPE = PREFIX + "eventType";
+
+	/**
+	 * The member header name.
+	 */
+	public static final String MEMBER = PREFIX + "member";
+
+	/**
+	 * The cache name header name.
+	 */
+	public static final String CACHE_NAME = PREFIX + "cacheName";
+
+	/**
+	 * The publishing time header name.
+	 */
+	public static final String PUBLISHING_TIME = PREFIX + "publishingTime";
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/HazelcastIntegrationDefinitionValidator.java

@@ -0,0 +1,108 @@
+/*
+ * 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;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.StringUtils;
+
+import com.hazelcast.collection.IList;
+import com.hazelcast.collection.IQueue;
+import com.hazelcast.collection.ISet;
+import com.hazelcast.core.DistributedObject;
+import com.hazelcast.map.IMap;
+import com.hazelcast.multimap.MultiMap;
+import com.hazelcast.replicatedmap.ReplicatedMap;
+import com.hazelcast.topic.ITopic;
+
+/**
+ * Common Validator for Hazelcast Integration. It validates cache types and events.
+ *
+ * @author Eren Avsarogullari
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+public final class HazelcastIntegrationDefinitionValidator {
+
+	public static <E extends Enum<E>> Set<String> validateEnumType(final Class<E> enumType, final String types) {
+			Set<String> typeSet = StringUtils.commaDelimitedListToSet(StringUtils.trimAllWhitespace(types));
+		for (String type : typeSet) {
+			Enum.valueOf(enumType, type);
+		}
+
+		return typeSet;
+	}
+
+	public static void validateCacheTypeForEventDrivenMessageProducer(final DistributedObject distributedObject) {
+		if (!(distributedObject instanceof IMap
+				|| distributedObject instanceof MultiMap
+				|| distributedObject instanceof ReplicatedMap
+				|| distributedObject instanceof IList
+				|| distributedObject instanceof ISet
+				|| distributedObject instanceof IQueue
+				|| distributedObject instanceof ITopic)) {
+			throw new IllegalArgumentException(
+					"Invalid 'cache' type is set. IMap, MultiMap, ReplicatedMap, IList, ISet, IQueue and ITopic" +
+							" cache object types are acceptable for Hazelcast Inbound Channel Adapter.");
+		}
+	}
+
+	public static void validateCacheEventsByDistributedObject(
+			final DistributedObject distributedObject, final Set<String> cacheEventTypeSet) {
+		List<String> supportedCacheEventTypes = getSupportedCacheEventTypes(distributedObject);
+		if (!CollectionUtils.isEmpty(supportedCacheEventTypes)) {
+			validateCacheEventsByDistributedObject(distributedObject, cacheEventTypeSet, supportedCacheEventTypes);
+		}
+	}
+
+	private static List<String> getSupportedCacheEventTypes(final DistributedObject distributedObject) {
+		if ((distributedObject instanceof IList)
+				|| (distributedObject instanceof ISet)
+				|| (distributedObject instanceof IQueue)) {
+			return Arrays.asList(CacheEventType.ADDED.toString(), CacheEventType.REMOVED.toString());
+		}
+		else if (distributedObject instanceof MultiMap) {
+			return Arrays.asList(CacheEventType.ADDED.toString(),
+					CacheEventType.REMOVED.toString(),
+					CacheEventType.CLEAR_ALL.toString());
+		}
+		else if (distributedObject instanceof ReplicatedMap) {
+			return Arrays.asList(CacheEventType.ADDED.toString(),
+					CacheEventType.REMOVED.toString(),
+					CacheEventType.UPDATED.toString(),
+					CacheEventType.EVICTED.toString());
+		}
+
+		return null;
+	}
+
+	private static void validateCacheEventsByDistributedObject(DistributedObject distributedObject,
+			Set<String> cacheEventTypeSet, List<String> supportedCacheEventTypes) {
+		if (!supportedCacheEventTypes.containsAll(cacheEventTypeSet)) {
+			throw new IllegalArgumentException("'cache-events' attribute of "
+					+ distributedObject.getName() + " can be set as " + supportedCacheEventTypes);
+		}
+	}
+
+	private HazelcastIntegrationDefinitionValidator() {
+	}
+
+}

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

@@ -0,0 +1,120 @@
+/*
+ * Copyright 2015-2019 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;
+
+import java.net.SocketAddress;
+import java.util.concurrent.locks.Lock;
+
+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;
+
+import com.hazelcast.cluster.MembershipListener;
+import com.hazelcast.core.Hazelcast;
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.multimap.MultiMap;
+
+/**
+ * This class creates an internal configuration {@link MultiMap} to cache Hazelcast instances' socket
+ * address information which used Hazelcast event-driven inbound channel adapter(s). It
+ * also enables a Hazelcast {@link MembershipListener} to listen for
+ * membership updates.
+ *
+ * @author Eren Avsarogullari
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+public class HazelcastLocalInstanceRegistrar implements SmartInitializingSingleton {
+
+	private static final Log logger = LogFactory.getLog(HazelcastLocalInstanceRegistrar.class);
+
+	/**
+	 * The bean name for the {@link HazelcastLocalInstanceRegistrar} instance.
+	 */
+	public static final String BEAN_NAME = "hazelcastLocalInstanceRegistrar";
+
+	/**
+	 * The name for the Hazelcast MultiMap used for membership registration.
+	 */
+	public static final String SPRING_INTEGRATION_INTERNAL_CLUSTER_MULTIMAP =
+			"SPRING_INTEGRATION_INTERNAL_CLUSTER_MULTIMAP";
+
+	/**
+	 * The name for the Hazelcast Lock used for membership registration.
+	 */
+	public static final String SPRING_INTEGRATION_INTERNAL_CLUSTER_LOCK = "SPRING_INTEGRATION_INTERNAL_CLUSTER_LOCK";
+
+	private final HazelcastInstance hazelcastInstance;
+
+	/**
+	 * Construct {@link HazelcastLocalInstanceRegistrar} based on the local JVM {@link HazelcastInstance}s if any.
+	 */
+	public HazelcastLocalInstanceRegistrar() {
+		this.hazelcastInstance = null;
+	}
+
+	/**
+	 * Construct {@link HazelcastLocalInstanceRegistrar} based on the provided {@link HazelcastInstance}.
+	 * @param hazelcastInstance the {@link HazelcastInstance} to use.
+	 */
+	public HazelcastLocalInstanceRegistrar(HazelcastInstance hazelcastInstance) {
+		this.hazelcastInstance = hazelcastInstance;
+	}
+
+	@Override
+	public void afterSingletonsInstantiated() {
+		if (this.hazelcastInstance == null) {
+			if (!Hazelcast.getAllHazelcastInstances().isEmpty()) {
+				HazelcastInstance hazelcastInstance = Hazelcast.getAllHazelcastInstances().iterator().next();
+				hazelcastInstance.getCluster().addMembershipListener(new HazelcastMembershipListener());
+				syncConfigurationMultiMap(hazelcastInstance);
+			}
+			else {
+				logger.warn("No HazelcastInstances for MembershipListener registration");
+			}
+		}
+		else {
+			syncConfigurationMultiMap(this.hazelcastInstance);
+			this.hazelcastInstance.getCluster().addMembershipListener(new HazelcastMembershipListener());
+		}
+	}
+
+	private void syncConfigurationMultiMap(HazelcastInstance hazelcastInstance) {
+		Lock lock = hazelcastInstance.getCPSubsystem().getLock(SPRING_INTEGRATION_INTERNAL_CLUSTER_LOCK);
+		lock.lock();
+		try {
+			MultiMap<SocketAddress, SocketAddress> multiMap = hazelcastInstance
+					.getMultiMap(SPRING_INTEGRATION_INTERNAL_CLUSTER_MULTIMAP);
+			for (HazelcastInstance localInstance : Hazelcast.getAllHazelcastInstances()) {
+				SocketAddress localInstanceSocketAddress = localInstance.getLocalEndpoint().getSocketAddress();
+				if (multiMap.size() == 0) {
+					multiMap.put(localInstanceSocketAddress, localInstanceSocketAddress);
+				}
+				else {
+					multiMap.put(multiMap.keySet().iterator().next(), localInstanceSocketAddress);
+				}
+			}
+		}
+		finally {
+			lock.unlock();
+		}
+	}
+
+}

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

@@ -0,0 +1,45 @@
+/*
+ * 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));
+		}
+	}
+
+}

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

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

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/xml/HazelcastClusterMonitorInboundChannelAdapterParser.java

@@ -0,0 +1,98 @@
+/*
+ * 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.xml;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.BeanDefinitionStoreException;
+import org.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
+import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.hazelcast.inbound.HazelcastClusterMonitorMessageProducer;
+import org.springframework.util.StringUtils;
+
+/**
+ * Parser for the {@code <int-hazelcast:cm-inbound-channel-adapter />} component.
+ *
+ * @author Eren Avsarogullari
+ * @since 6.0
+ */
+public class HazelcastClusterMonitorInboundChannelAdapterParser extends
+		AbstractSingleBeanDefinitionParser {
+
+	private static final String CHANNEL_ATTRIBUTE = "channel";
+
+	private static final String HAZELCAST_INSTANCE_ATTRIBUTE = "hazelcast-instance";
+
+	private static final String MONITOR_TYPES_ATTRIBUTE = "monitor-types";
+
+	private static final String OUTPUT_CHANNEL = "outputChannel";
+
+	private static final String MONITOR_EVENT_TYPES = "monitorEventTypes";
+
+	@Override
+	protected Class<?> getBeanClass(Element element) {
+		return HazelcastClusterMonitorMessageProducer.class;
+	}
+
+	@Override
+	protected String resolveId(Element element, AbstractBeanDefinition definition,
+			ParserContext parserContext) throws BeanDefinitionStoreException {
+		String id = super.resolveId(element, definition, parserContext);
+
+		if (!element.hasAttribute(CHANNEL_ATTRIBUTE)) {
+			id = id + ".adapter";
+		}
+
+		if (!StringUtils.hasText(id)) {
+			id = BeanDefinitionReaderUtils.generateBeanName(definition,
+					parserContext.getRegistry());
+		}
+
+		return id;
+	}
+
+	@Override
+	protected void doParse(Element element, ParserContext parserContext,
+			BeanDefinitionBuilder builder) {
+		String channelName = element.getAttribute(CHANNEL_ATTRIBUTE);
+		if (!StringUtils.hasText(channelName)) {
+			channelName = IntegrationNamespaceUtils.createDirectChannel(element,
+					parserContext);
+		}
+
+		if (!StringUtils.hasText(element.getAttribute(HAZELCAST_INSTANCE_ATTRIBUTE))) {
+			parserContext.getReaderContext().error(
+					"'" + HAZELCAST_INSTANCE_ATTRIBUTE + "' attribute is required.",
+					element);
+		}
+
+		builder.addPropertyReference(OUTPUT_CHANNEL, channelName);
+		builder.addConstructorArgReference(element
+				.getAttribute(HAZELCAST_INSTANCE_ATTRIBUTE));
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
+				MONITOR_TYPES_ATTRIBUTE, MONITOR_EVENT_TYPES);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
+				IntegrationNamespaceUtils.AUTO_STARTUP);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
+				IntegrationNamespaceUtils.PHASE);
+	}
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/xml/HazelcastContinuousQueryInboundChannelAdapterParser.java

@@ -0,0 +1,108 @@
+/*
+ * Copyright 2015 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.xml;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.BeanDefinitionStoreException;
+import org.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
+import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.hazelcast.inbound.HazelcastContinuousQueryMessageProducer;
+import org.springframework.util.StringUtils;
+
+/**
+ * Hazelcast Continuous Query Inbound Channel Adapter Parser parses
+ * {@code <int-hazelcast:cq-inbound-channel-adapter/>} configuration.
+ *
+ * @author Eren Avsarogullari
+ * @since 6.0
+ */
+public class HazelcastContinuousQueryInboundChannelAdapterParser extends AbstractSingleBeanDefinitionParser {
+
+	private static final String CHANNEL_ATTRIBUTE = "channel";
+
+	private static final String CACHE_ATTRIBUTE = "cache";
+
+	private static final String CACHE_EVENTS_ATTRIBUTE = "cache-events";
+
+	private static final String PREDICATE_ATTRIBUTE = "predicate";
+
+	private static final String INCLUDE_VALUE_ATTRIBUTE = "include-value";
+
+	private static final String CACHE_LISTENING_POLICY_ATTRIBUTE = "cache-listening-policy";
+
+	private static final String OUTPUT_CHANNEL = "outputChannel";
+
+	private static final String CACHE_EVENT_TYPES = "cacheEventTypes";
+
+	@Override
+	protected Class<?> getBeanClass(Element element) {
+		return HazelcastContinuousQueryMessageProducer.class;
+	}
+
+	@Override
+	protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext)
+			throws BeanDefinitionStoreException {
+		String id = super.resolveId(element, definition, parserContext);
+
+		if (!element.hasAttribute(CHANNEL_ATTRIBUTE)) {
+			id = id + ".adapter";
+		}
+
+		if (!StringUtils.hasText(id)) {
+			id = BeanDefinitionReaderUtils.generateBeanName(definition, parserContext.getRegistry());
+		}
+
+		return id;
+	}
+
+	@Override
+	protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
+		String channelName = element.getAttribute(CHANNEL_ATTRIBUTE);
+		if (!StringUtils.hasText(channelName)) {
+			channelName = IntegrationNamespaceUtils.createDirectChannel(element, parserContext);
+		}
+
+		if (!StringUtils.hasText(element.getAttribute(CACHE_ATTRIBUTE))) {
+			parserContext.getReaderContext().error("'" + CACHE_ATTRIBUTE + "' attribute is required.", element);
+		}
+		else if (!StringUtils.hasText(element.getAttribute(CACHE_EVENTS_ATTRIBUTE))) {
+			parserContext.getReaderContext().error("'" + CACHE_EVENTS_ATTRIBUTE + "' attribute is required.", element);
+		}
+		else if (!StringUtils.hasText(element.getAttribute(PREDICATE_ATTRIBUTE))) {
+			parserContext.getReaderContext().error("'" + PREDICATE_ATTRIBUTE + "' attribute is required.", element);
+		}
+		else if (!StringUtils.hasText(element.getAttribute(CACHE_LISTENING_POLICY_ATTRIBUTE))) {
+			parserContext.getReaderContext().error("'" + CACHE_LISTENING_POLICY_ATTRIBUTE + "' attribute is required.",
+					element);
+		}
+
+		builder.addPropertyReference(OUTPUT_CHANNEL, channelName);
+		builder.addConstructorArgReference(element.getAttribute(CACHE_ATTRIBUTE));
+		builder.addConstructorArgValue(element.getAttribute(PREDICATE_ATTRIBUTE));
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, CACHE_EVENTS_ATTRIBUTE, CACHE_EVENT_TYPES);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, INCLUDE_VALUE_ATTRIBUTE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, CACHE_LISTENING_POLICY_ATTRIBUTE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.AUTO_STARTUP);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.PHASE);
+	}
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/xml/HazelcastDistributedSQLInboundChannelAdapterParser.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2015-2016 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.xml;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.BeanMetadataElement;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.hazelcast.inbound.HazelcastDistributedSQLMessageSource;
+import org.springframework.util.StringUtils;
+
+/**
+ * Hazelcast Distributed SQL Inbound Channel Adapter Parser parses
+ * {@code <int-hazelcast:ds-inbound-channel-adapter/>} configuration.
+ *
+ * @author Eren Avsarogullari
+ * @since 6.0
+ */
+public class HazelcastDistributedSQLInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
+
+	private static final String CACHE_ATTRIBUTE = "cache";
+
+	private static final String DISTRIBUTED_SQL_ATTRIBUTE = "distributed-sql";
+
+	private static final String ITERATION_TYPE_ATTRIBUTE = "iteration-type";
+
+	@Override
+	protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
+		if (!StringUtils.hasText(element.getAttribute(CACHE_ATTRIBUTE))) {
+			parserContext.getReaderContext().error("'" + CACHE_ATTRIBUTE + "' attribute is required.", element);
+		}
+		else if (!StringUtils.hasText(element.getAttribute(DISTRIBUTED_SQL_ATTRIBUTE))) {
+			parserContext.getReaderContext().error("'" + DISTRIBUTED_SQL_ATTRIBUTE + "' attribute is required.",
+					element);
+		}
+		else if (!StringUtils.hasText(element.getAttribute(ITERATION_TYPE_ATTRIBUTE))) {
+			parserContext.getReaderContext().error("'" + ITERATION_TYPE_ATTRIBUTE + "' attribute is required.",
+					element);
+		}
+
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder
+				.genericBeanDefinition(HazelcastDistributedSQLMessageSource.class.getName());
+
+		builder.addConstructorArgReference(element.getAttribute(CACHE_ATTRIBUTE));
+		builder.addConstructorArgValue(element.getAttribute(DISTRIBUTED_SQL_ATTRIBUTE));
+
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, ITERATION_TYPE_ATTRIBUTE);
+
+		return builder.getBeanDefinition();
+	}
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/xml/HazelcastEventDrivenInboundChannelAdapterParser.java

@@ -0,0 +1,99 @@
+/*
+ * Copyright 2015 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.xml;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.BeanDefinitionStoreException;
+import org.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
+import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.hazelcast.inbound.HazelcastEventDrivenMessageProducer;
+import org.springframework.util.StringUtils;
+
+/**
+ * Hazelcast Event Driven Inbound Channel Adapter Parser parses
+ * {@code <int-hazelcast:inbound-channel-adapter />} configuration.
+ *
+ * @author Eren Avsarogullari
+ * @since 6.0
+ */
+public class HazelcastEventDrivenInboundChannelAdapterParser extends AbstractSingleBeanDefinitionParser {
+
+	private static final String CHANNEL_ATTRIBUTE = "channel";
+
+	private static final String CACHE_ATTRIBUTE = "cache";
+
+	private static final String CACHE_EVENTS_ATTRIBUTE = "cache-events";
+
+	private static final String CACHE_LISTENING_POLICY_ATTRIBUTE = "cache-listening-policy";
+
+	private static final String OUTPUT_CHANNEL = "outputChannel";
+
+	private static final String CACHE_EVENT_TYPES = "cacheEventTypes";
+
+	@Override
+	protected Class<?> getBeanClass(Element element) {
+		return HazelcastEventDrivenMessageProducer.class;
+	}
+
+	@Override
+	protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext)
+			throws BeanDefinitionStoreException {
+		String id = super.resolveId(element, definition, parserContext);
+
+		if (!element.hasAttribute(CHANNEL_ATTRIBUTE)) {
+			id = id + ".adapter";
+		}
+
+		if (!StringUtils.hasText(id)) {
+			id = BeanDefinitionReaderUtils.generateBeanName(definition, parserContext.getRegistry());
+		}
+
+		return id;
+	}
+
+	@Override
+	protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
+		String channelName = element.getAttribute(CHANNEL_ATTRIBUTE);
+		if (!StringUtils.hasText(channelName)) {
+			channelName = IntegrationNamespaceUtils.createDirectChannel(element, parserContext);
+		}
+
+		if (!StringUtils.hasText(element.getAttribute(CACHE_ATTRIBUTE))) {
+			parserContext.getReaderContext().error("'" + CACHE_ATTRIBUTE + "' attribute is required.", element);
+		}
+		else if (!StringUtils.hasText(element.getAttribute(CACHE_EVENTS_ATTRIBUTE))) {
+			parserContext.getReaderContext().error("'" + CACHE_EVENTS_ATTRIBUTE + "' attribute is required.", element);
+		}
+		else if (!StringUtils.hasText(element.getAttribute(CACHE_LISTENING_POLICY_ATTRIBUTE))) {
+			parserContext.getReaderContext().error("'" + CACHE_LISTENING_POLICY_ATTRIBUTE + "' attribute is required.",
+					element);
+		}
+
+		builder.addPropertyReference(OUTPUT_CHANNEL, channelName);
+		builder.addConstructorArgReference(element.getAttribute(CACHE_ATTRIBUTE));
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, CACHE_EVENTS_ATTRIBUTE, CACHE_EVENT_TYPES);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, CACHE_LISTENING_POLICY_ATTRIBUTE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.AUTO_STARTUP);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.PHASE);
+	}
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/xml/HazelcastIntegrationNamespaceHandler.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2015 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.xml;
+
+import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler;
+
+/**
+ * Namespace handler for the Hazelcast schema.
+ *
+ * @author Eren Avsarogullari
+ * @since 6.0
+ */
+public class HazelcastIntegrationNamespaceHandler extends AbstractIntegrationNamespaceHandler {
+
+	@Override
+	public void init() {
+		registerBeanDefinitionParser("inbound-channel-adapter", new HazelcastEventDrivenInboundChannelAdapterParser());
+		registerBeanDefinitionParser("outbound-channel-adapter", new HazelcastOutboundChannelAdapterParser());
+		registerBeanDefinitionParser("cq-inbound-channel-adapter", new HazelcastContinuousQueryInboundChannelAdapterParser());
+		registerBeanDefinitionParser("ds-inbound-channel-adapter", new HazelcastDistributedSQLInboundChannelAdapterParser());
+		registerBeanDefinitionParser("cm-inbound-channel-adapter", new HazelcastClusterMonitorInboundChannelAdapterParser());
+	}
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/xml/HazelcastOutboundChannelAdapterParser.java

@@ -0,0 +1,71 @@
+/*
+ * Copyright 2015 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.xml;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.hazelcast.outbound.HazelcastCacheWritingMessageHandler;
+
+/**
+ * Hazelcast Outbound Channel Adapter Parser for
+ * {@code <int-hazelcast:inbound-channel-adapter />}.
+ *
+ * @author Eren Avsarogullari
+ * @since 6.0
+ */
+public class HazelcastOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
+
+	private static final String CACHE_ATTRIBUTE = "cache";
+
+	private static final String CACHE_EXPRESSION_ATTRIBUTE = "cache-expression";
+
+	private static final String KEY_EXPRESSION_ATTRIBUTE = "key-expression";
+
+	private static final String EXTRACT_PAYLOAD_ATTRIBUTE = "extract-payload";
+
+	private static final String DISTRIBUTED_OBJECT = "distributedObject";
+
+	@Override
+	protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder
+				.genericBeanDefinition(HazelcastCacheWritingMessageHandler.class);
+
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, CACHE_ATTRIBUTE, DISTRIBUTED_OBJECT);
+		BeanDefinition cacheExpressionDef =
+				IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined(CACHE_EXPRESSION_ATTRIBUTE, element);
+		if (cacheExpressionDef != null) {
+			builder.addPropertyValue("cacheExpression", cacheExpressionDef);
+		}
+
+		BeanDefinition keyExpressionDef =
+				IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined(KEY_EXPRESSION_ATTRIBUTE, element);
+		if (keyExpressionDef != null) {
+			builder.addPropertyValue("keyExpression", keyExpressionDef);
+		}
+
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, EXTRACT_PAYLOAD_ATTRIBUTE);
+
+		return builder.getBeanDefinition();
+	}
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/config/xml/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides classes for parsers and namespace handlers.
+ */
+package org.springframework.integration.hazelcast.config.xml;

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

@@ -0,0 +1,232 @@
+/*
+ * 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.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;
+
+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;
+
+import com.hazelcast.core.DistributedObject;
+import com.hazelcast.core.EntryEvent;
+import com.hazelcast.core.EntryListener;
+import com.hazelcast.core.Hazelcast;
+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;
+
+/**
+ * Hazelcast Base Event-Driven Message Producer.
+ *
+ * @author Eren Avsarogullari
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+public abstract class AbstractHazelcastMessageProducer extends MessageProducerSupport {
+
+	protected final DistributedObject distributedObject;
+
+	private volatile CacheListeningPolicyType cacheListeningPolicy = CacheListeningPolicyType.SINGLE;
+
+	private volatile UUID hazelcastRegisteredEventListenerId;
+
+	private Set<String> cacheEvents = Collections.singleton(CacheEventType.ADDED.name());
+
+	public AbstractHazelcastMessageProducer(DistributedObject distributedObject) {
+		Assert.notNull(distributedObject, "'distributedObject' must not be null");
+		this.distributedObject = distributedObject;
+	}
+
+	protected Set<String> getCacheEvents() {
+		return this.cacheEvents;
+	}
+
+	public void setCacheEventTypes(String cacheEventTypes) {
+		Set<String> cacheEvents =
+				HazelcastIntegrationDefinitionValidator.validateEnumType(CacheEventType.class, cacheEventTypes);
+		Assert.notEmpty(cacheEvents, "'cacheEvents' must have elements");
+		HazelcastIntegrationDefinitionValidator.validateCacheEventsByDistributedObject(this.distributedObject,
+				cacheEvents);
+		this.cacheEvents = cacheEvents;
+	}
+
+	protected CacheListeningPolicyType getCacheListeningPolicy() {
+		return this.cacheListeningPolicy;
+	}
+
+	public void setCacheListeningPolicy(CacheListeningPolicyType cacheListeningPolicy) {
+		Assert.notNull(cacheListeningPolicy, "'cacheListeningPolicy' must not be null");
+		this.cacheListeningPolicy = cacheListeningPolicy;
+	}
+
+	protected UUID getHazelcastRegisteredEventListenerId() {
+		return this.hazelcastRegisteredEventListenerId;
+	}
+
+	protected void setHazelcastRegisteredEventListenerId(UUID hazelcastRegisteredEventListenerId) {
+		this.hazelcastRegisteredEventListenerId = hazelcastRegisteredEventListenerId;
+	}
+
+	/**
+	 * A base event listener abstraction.
+	 *
+	 * @param <E> the event type
+	 */
+	protected abstract class AbstractHazelcastEventListener<E> {
+
+		protected abstract void processEvent(E event);
+
+		protected abstract Message<?> toMessage(E event);
+
+		protected void sendMessage(E event, InetSocketAddress socketAddress,
+				CacheListeningPolicyType cacheListeningPolicyType) {
+			if (CacheListeningPolicyType.ALL == cacheListeningPolicyType || isEventAcceptable(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<>();
+			for (HazelcastInstance hazelcastInstance : hazelcastInstanceSet) {
+				localSocketAddressesSet.add(hazelcastInstance.getLocalEndpoint().getSocketAddress());
+			}
+
+			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());
+
+		}
+
+	}
+
+	/**
+	 * The {@link AbstractHazelcastEventListener} implementation for the {@link AbstractIMapEvent}s.
+	 *
+	 * @param <K> the entry key type
+	 * @param <V> the entry value type
+	 */
+	protected final class HazelcastEntryListener<K, V> extends
+			AbstractHazelcastEventListener<AbstractIMapEvent> implements EntryListener<K, V> {
+
+		@Override
+		public void entryAdded(EntryEvent<K, V> event) {
+			processEvent(event);
+		}
+
+		@Override
+		public void entryRemoved(EntryEvent<K, V> event) {
+			processEvent(event);
+		}
+
+		@Override
+		public void entryUpdated(EntryEvent<K, V> event) {
+			processEvent(event);
+		}
+
+		@Override
+		public void entryEvicted(EntryEvent<K, V> event) {
+			processEvent(event);
+		}
+
+		@Override
+		public void entryExpired(EntryEvent<K, V> event) {
+			processEvent(event);
+		}
+
+		@Override
+		public void mapEvicted(MapEvent event) {
+			processEvent(event);
+		}
+
+		@Override
+		public void mapCleared(MapEvent event) {
+			processEvent(event);
+		}
+
+		@Override
+		protected void processEvent(AbstractIMapEvent event) {
+			if (getCacheEvents().contains(event.getEventType().toString())) {
+				if (AbstractHazelcastMessageProducer.this.logger.isDebugEnabled()) {
+					AbstractHazelcastMessageProducer.this.logger.debug("Received Event : " + event);
+				}
+				sendMessage(event,
+						event.getMember().getSocketAddress(EndpointQualifier.MEMBER), getCacheListeningPolicy());
+			}
+		}
+
+		@Override
+		protected Message<?> toMessage(AbstractIMapEvent event) {
+			final Map<String, Object> headers = new HashMap<>();
+			headers.put(HazelcastHeaders.EVENT_TYPE, event.getEventType().name());
+			headers.put(HazelcastHeaders.MEMBER, event.getMember().getSocketAddress(EndpointQualifier.MEMBER));
+			headers.put(HazelcastHeaders.CACHE_NAME, event.getName());
+
+			if (event instanceof EntryEvent) {
+				@SuppressWarnings("unchecked")
+				EntryEvent<K, V> entryEvent = (EntryEvent<K, V>) event;
+				EntryEventMessagePayload<K, V> messagePayload = new EntryEventMessagePayload<>(entryEvent.getKey(),
+						entryEvent.getValue(), entryEvent.getOldValue());
+				return getMessageBuilderFactory().withPayload(messagePayload).copyHeaders(headers).build();
+			}
+			else if (event instanceof MapEvent) {
+				return getMessageBuilderFactory()
+						.withPayload(((MapEvent) event).getNumberOfEntriesAffected()).copyHeaders(headers).build();
+			}
+			else {
+				throw new IllegalStateException("Invalid event is received. Event : " + event);
+			}
+		}
+
+	}
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/HazelcastClusterMonitorMessageProducer.java

@@ -0,0 +1,213 @@
+/*
+ * Copyright 2015 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.inbound;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.integration.hazelcast.ClusterMonitorType;
+import org.springframework.integration.hazelcast.HazelcastIntegrationDefinitionValidator;
+import org.springframework.util.Assert;
+
+import com.hazelcast.client.Client;
+import com.hazelcast.client.ClientListener;
+import com.hazelcast.cluster.MembershipEvent;
+import com.hazelcast.cluster.MembershipListener;
+import com.hazelcast.core.DistributedObjectEvent;
+import com.hazelcast.core.DistributedObjectListener;
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.core.LifecycleEvent;
+import com.hazelcast.core.LifecycleListener;
+import com.hazelcast.partition.MigrationListener;
+import com.hazelcast.partition.MigrationState;
+import com.hazelcast.partition.ReplicaMigrationEvent;
+
+/**
+ * Hazelcast Cluster Monitor Event Driven Message Producer is a message producer which
+ * enables {@link HazelcastClusterMonitorMessageProducer.HazelcastClusterMonitorListener}
+ * listener in order to listen cluster related events and sends events to related channel.
+ *
+ * @author Eren Avsarogullari
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+public class HazelcastClusterMonitorMessageProducer extends MessageProducerSupport {
+
+	private final HazelcastInstance hazelcastInstance;
+
+	private Set<String> monitorTypes = Collections.singleton(ClusterMonitorType.MEMBERSHIP.name());
+
+	private final Map<ClusterMonitorType, UUID> hazelcastRegisteredListenerIdMap = new ConcurrentHashMap<>(5);
+
+	public HazelcastClusterMonitorMessageProducer(HazelcastInstance hazelcastInstance) {
+		Assert.notNull(hazelcastInstance, "'hazelcastInstance' must not be null");
+		this.hazelcastInstance = hazelcastInstance;
+	}
+
+	public void setMonitorEventTypes(String monitorEventTypes) {
+		final Set<String> monitorTypes =
+				HazelcastIntegrationDefinitionValidator.validateEnumType(ClusterMonitorType.class, monitorEventTypes);
+		Assert.notEmpty(monitorTypes, "'monitorTypes' must have elements");
+		this.monitorTypes = monitorTypes;
+	}
+
+	@Override
+	protected void doStart() {
+		final HazelcastClusterMonitorListener clusterMonitorListener = new HazelcastClusterMonitorListener();
+
+		if (this.monitorTypes.contains(ClusterMonitorType.MEMBERSHIP.name())) {
+			final UUID registrationId = this.hazelcastInstance.getCluster()
+					.addMembershipListener(clusterMonitorListener);
+			this.hazelcastRegisteredListenerIdMap.put(ClusterMonitorType.MEMBERSHIP, registrationId);
+		}
+
+		if (this.monitorTypes.contains(ClusterMonitorType.DISTRIBUTED_OBJECT.name())) {
+			final UUID registrationId = this.hazelcastInstance
+					.addDistributedObjectListener(clusterMonitorListener);
+			this.hazelcastRegisteredListenerIdMap.put(ClusterMonitorType.DISTRIBUTED_OBJECT, registrationId);
+		}
+
+		if (this.monitorTypes.contains(ClusterMonitorType.MIGRATION.name())) {
+			final UUID registrationId = this.hazelcastInstance.getPartitionService()
+					.addMigrationListener(clusterMonitorListener);
+			this.hazelcastRegisteredListenerIdMap.put(ClusterMonitorType.MIGRATION,	registrationId);
+		}
+
+		if (this.monitorTypes.contains(ClusterMonitorType.LIFECYCLE.name())) {
+			final UUID registrationId = this.hazelcastInstance.getLifecycleService()
+					.addLifecycleListener(clusterMonitorListener);
+			this.hazelcastRegisteredListenerIdMap.put(ClusterMonitorType.LIFECYCLE,	registrationId);
+		}
+
+		if (this.monitorTypes.contains(ClusterMonitorType.CLIENT.name())) {
+			final UUID registrationId = this.hazelcastInstance.getClientService()
+					.addClientListener(clusterMonitorListener);
+			this.hazelcastRegisteredListenerIdMap.put(ClusterMonitorType.CLIENT, registrationId);
+		}
+	}
+
+	@Override
+	protected void doStop() {
+		if (this.hazelcastInstance.getLifecycleService().isRunning()) {
+			UUID id = this.hazelcastRegisteredListenerIdMap.remove(ClusterMonitorType.MEMBERSHIP);
+			if (id != null) {
+				this.hazelcastInstance.getCluster().removeMembershipListener(id);
+			}
+
+			id = this.hazelcastRegisteredListenerIdMap.remove(ClusterMonitorType.DISTRIBUTED_OBJECT);
+			if (id != null) {
+				this.hazelcastInstance.removeDistributedObjectListener(id);
+			}
+
+			id = this.hazelcastRegisteredListenerIdMap.remove(ClusterMonitorType.MIGRATION);
+			if (id != null) {
+				this.hazelcastInstance.getPartitionService().removeMigrationListener(id);
+			}
+
+			id = this.hazelcastRegisteredListenerIdMap.remove(ClusterMonitorType.LIFECYCLE);
+			if (id != null) {
+				this.hazelcastInstance.getLifecycleService().removeLifecycleListener(id);
+			}
+
+			id = this.hazelcastRegisteredListenerIdMap.remove(ClusterMonitorType.CLIENT);
+			if (id != null) {
+				this.hazelcastInstance.getClientService().removeClientListener(id);
+			}
+		}
+	}
+
+	@Override
+	public String getComponentType() {
+		return "hazelcast:cm-inbound-channel-adapter";
+	}
+
+	private void processEvent(Object event) {
+		Assert.notNull(event, "'hazelcast event' must not be null");
+		if (logger.isDebugEnabled()) {
+			logger.debug("Received Cluster Monitor Event : " + event);
+		}
+		this.sendMessage(getMessageBuilderFactory().withPayload(event).build());
+
+	}
+
+	private final class HazelcastClusterMonitorListener implements MembershipListener,
+			DistributedObjectListener, MigrationListener, LifecycleListener,
+			ClientListener {
+
+		@Override
+		public void memberAdded(MembershipEvent membershipEvent) {
+			processEvent(membershipEvent);
+		}
+
+		@Override
+		public void memberRemoved(MembershipEvent membershipEvent) {
+			processEvent(membershipEvent);
+		}
+
+		@Override
+		public void distributedObjectCreated(DistributedObjectEvent event) {
+			processEvent(event);
+		}
+
+		@Override
+		public void distributedObjectDestroyed(DistributedObjectEvent event) {
+			processEvent(event);
+		}
+
+		@Override
+		public void migrationStarted(MigrationState migrationEvent) {
+			processEvent(migrationEvent);
+		}
+
+		@Override
+		public void migrationFinished(MigrationState migrationEvent) {
+			processEvent(migrationEvent);
+		}
+
+		@Override
+		public void replicaMigrationCompleted(ReplicaMigrationEvent event) {
+			processEvent(event);
+		}
+
+		@Override
+		public void replicaMigrationFailed(ReplicaMigrationEvent event) {
+			processEvent(event);
+		}
+
+		@Override
+		public void stateChanged(LifecycleEvent event) {
+			processEvent(event);
+		}
+
+		@Override
+		public void clientConnected(Client client) {
+			processEvent(client);
+		}
+
+		@Override
+		public void clientDisconnected(Client client) {
+			processEvent(client);
+		}
+
+	}
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/HazelcastContinuousQueryMessageProducer.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2015 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.inbound;
+
+import org.springframework.util.Assert;
+
+import com.hazelcast.map.IMap;
+import com.hazelcast.query.impl.predicates.SqlPredicate;
+
+/**
+ * Hazelcast Continuous Query Message Producer is a message producer which enables
+ * {@link AbstractHazelcastMessageProducer.HazelcastEntryListener} with a
+ * {@link SqlPredicate} in order to listen related distributed map events in the light of
+ * defined predicate and sends events to related channel.
+ *
+ * @author Eren Avsarogullari
+ * @since 6.0
+ */
+public class HazelcastContinuousQueryMessageProducer extends AbstractHazelcastMessageProducer {
+
+	private final String predicate;
+
+	private boolean includeValue = true;
+
+	@SuppressWarnings("rawtypes")
+	public HazelcastContinuousQueryMessageProducer(IMap distributedMap, String predicate) {
+		super(distributedMap);
+		Assert.hasText(predicate, "'predicate' must not be null");
+		this.predicate = predicate;
+	}
+
+	public void setIncludeValue(boolean includeValue) {
+		this.includeValue = includeValue;
+	}
+
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	@Override
+	protected void doStart() {
+		setHazelcastRegisteredEventListenerId(((IMap<?, ?>) this.distributedObject)
+				.addEntryListener(new HazelcastEntryListener(), new SqlPredicate(this.predicate),
+						this.includeValue));
+	}
+
+	@Override
+	protected void doStop() {
+		((IMap<?, ?>) this.distributedObject).removeEntryListener(getHazelcastRegisteredEventListenerId());
+	}
+
+	@Override
+	public String getComponentType() {
+		return "hazelcast:cq-inbound-channel-adapter";
+	}
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/HazelcastDistributedSQLMessageSource.java

@@ -0,0 +1,92 @@
+/*
+ * Copyright 2015 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.inbound;
+
+import java.util.Collection;
+import java.util.Collections;
+
+import org.springframework.integration.endpoint.AbstractMessageSource;
+import org.springframework.integration.hazelcast.DistributedSQLIterationType;
+import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+
+import com.hazelcast.map.IMap;
+import com.hazelcast.query.impl.predicates.SqlPredicate;
+
+/**
+ * Hazelcast Distributed SQL Message Source is a message source which runs defined
+ * distributed query in the cluster and returns results in the light of iteration type.
+ *
+ * @author Eren Avsarogullari
+ * @since 6.0
+ */
+@SuppressWarnings("rawtypes")
+public class HazelcastDistributedSQLMessageSource extends AbstractMessageSource {
+
+	private final IMap<?, ?> distributedMap;
+
+	private final String distributedSql;
+
+	private DistributedSQLIterationType iterationType = DistributedSQLIterationType.VALUE;
+
+	public HazelcastDistributedSQLMessageSource(IMap distributedMap, String distributedSql) {
+		Assert.notNull(distributedMap, "'distributedMap' must not be null");
+		Assert.hasText(distributedSql, "'distributedSql' must not be empty");
+		this.distributedMap = distributedMap;
+		this.distributedSql = distributedSql;
+	}
+
+	public void setIterationType(DistributedSQLIterationType iterationType) {
+		Assert.notNull(this.iterationType, "'iterationType' must not be null");
+		this.iterationType = iterationType;
+	}
+
+	@Override
+	public String getComponentType() {
+		return "hazelcast:ds-inbound-channel-adapter";
+	}
+
+	@Override
+	@SuppressWarnings("unchecked")
+	protected Collection<?> doReceive() {
+		switch (this.iterationType) {
+			case ENTRY:
+				return getDistributedSQLResultSet(Collections
+						.unmodifiableCollection(this.distributedMap.entrySet(new SqlPredicate(this.distributedSql))));
+
+			case KEY:
+				return getDistributedSQLResultSet(Collections
+						.unmodifiableCollection(this.distributedMap.keySet(new SqlPredicate(this.distributedSql))));
+
+			case LOCAL_KEY:
+				return getDistributedSQLResultSet(Collections
+						.unmodifiableCollection(this.distributedMap.localKeySet(new SqlPredicate(this.distributedSql))));
+
+			default:
+				return getDistributedSQLResultSet(this.distributedMap.values(new SqlPredicate(this.distributedSql)));
+		}
+	}
+
+	private Collection<?> getDistributedSQLResultSet(Collection<?> collection) {
+		if (CollectionUtils.isEmpty(collection)) {
+			return null;
+		}
+
+		return collection;
+	}
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/HazelcastEventDrivenMessageProducer.java

@@ -0,0 +1,197 @@
+/*
+ * Copyright 2015-2019 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.inbound;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.integration.hazelcast.HazelcastHeaders;
+import org.springframework.integration.hazelcast.HazelcastIntegrationDefinitionValidator;
+import org.springframework.util.Assert;
+
+import com.hazelcast.collection.IList;
+import com.hazelcast.collection.IQueue;
+import com.hazelcast.collection.ISet;
+import com.hazelcast.collection.ItemEvent;
+import com.hazelcast.collection.ItemListener;
+import com.hazelcast.core.DistributedObject;
+import com.hazelcast.instance.EndpointQualifier;
+import com.hazelcast.map.IMap;
+import com.hazelcast.map.listener.MapListener;
+import com.hazelcast.multimap.MultiMap;
+import com.hazelcast.replicatedmap.ReplicatedMap;
+import com.hazelcast.topic.ITopic;
+import com.hazelcast.topic.Message;
+import com.hazelcast.topic.MessageListener;
+
+/**
+ * Hazelcast Event Driven Message Producer is a message producer which enables
+ * {@link AbstractHazelcastMessageProducer.HazelcastEntryListener},
+ * {@link HazelcastEventDrivenMessageProducer.HazelcastItemListener} and
+ * {@link HazelcastEventDrivenMessageProducer.HazelcastMessageListener} listeners in order
+ * to listen related cache events and sends events to related channel.
+ *
+ * @author Eren Avsarogullari
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+public class HazelcastEventDrivenMessageProducer extends AbstractHazelcastMessageProducer {
+
+	public HazelcastEventDrivenMessageProducer(DistributedObject distributedObject) {
+		super(distributedObject);
+	}
+
+	@Override
+	protected void onInit() {
+		super.onInit();
+		HazelcastIntegrationDefinitionValidator.validateCacheTypeForEventDrivenMessageProducer(this.distributedObject);
+	}
+
+	@Override
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	protected void doStart() {
+		if (this.distributedObject instanceof IMap) {
+			setHazelcastRegisteredEventListenerId(((IMap<?, ?>) this.distributedObject)
+					.addEntryListener((MapListener) new HazelcastEntryListener(), true));
+		}
+		else if (this.distributedObject instanceof MultiMap) {
+			setHazelcastRegisteredEventListenerId(((MultiMap<?, ?>) this.distributedObject)
+					.addEntryListener(new HazelcastEntryListener(), true));
+		}
+		else if (this.distributedObject instanceof ReplicatedMap) {
+			setHazelcastRegisteredEventListenerId(((ReplicatedMap<?, ?>) this.distributedObject)
+					.addEntryListener(new HazelcastEntryListener()));
+		}
+		else if (this.distributedObject instanceof IList) {
+			setHazelcastRegisteredEventListenerId(((IList<?>) this.distributedObject)
+					.addItemListener(new HazelcastItemListener(), true));
+		}
+		else if (this.distributedObject instanceof ISet) {
+			setHazelcastRegisteredEventListenerId(((ISet<?>) this.distributedObject)
+					.addItemListener(new HazelcastItemListener(), true));
+		}
+		else if (this.distributedObject instanceof IQueue) {
+			setHazelcastRegisteredEventListenerId(((IQueue<?>) this.distributedObject)
+					.addItemListener(new HazelcastItemListener(), true));
+		}
+		else if (this.distributedObject instanceof ITopic) {
+			setHazelcastRegisteredEventListenerId(((ITopic<?>) this.distributedObject)
+					.addMessageListener(new HazelcastMessageListener()));
+		}
+	}
+
+	@Override
+	protected void doStop() {
+		if (this.distributedObject instanceof IMap) {
+			((IMap<?, ?>) this.distributedObject).removeEntryListener(getHazelcastRegisteredEventListenerId());
+		}
+		else if (this.distributedObject instanceof MultiMap) {
+			((MultiMap<?, ?>) this.distributedObject).removeEntryListener(getHazelcastRegisteredEventListenerId());
+		}
+		else if (this.distributedObject instanceof ReplicatedMap) {
+			((ReplicatedMap<?, ?>) this.distributedObject).removeEntryListener(getHazelcastRegisteredEventListenerId());
+		}
+		else if (this.distributedObject instanceof IList) {
+			((IList<?>) this.distributedObject).removeItemListener(getHazelcastRegisteredEventListenerId());
+		}
+		else if (this.distributedObject instanceof ISet) {
+			((ISet<?>) this.distributedObject).removeItemListener(getHazelcastRegisteredEventListenerId());
+		}
+		else if (this.distributedObject instanceof IQueue) {
+			((IQueue<?>) this.distributedObject).removeItemListener(getHazelcastRegisteredEventListenerId());
+		}
+		else if (this.distributedObject instanceof ITopic) {
+			((ITopic<?>) this.distributedObject).removeMessageListener(getHazelcastRegisteredEventListenerId());
+		}
+	}
+
+	@Override
+	public String getComponentType() {
+		return "hazelcast:inbound-channel-adapter";
+	}
+
+	private class HazelcastItemListener<E> extends AbstractHazelcastEventListener<ItemEvent<E>>
+			implements ItemListener<E> {
+
+		@Override
+		public void itemAdded(ItemEvent<E> item) {
+			processEvent(item);
+		}
+
+		@Override
+		public void itemRemoved(ItemEvent<E> item) {
+			processEvent(item);
+		}
+
+		@Override
+		protected void processEvent(ItemEvent<E> event) {
+			if (getCacheEvents().contains(event.getEventType().toString())) {
+				sendMessage(event,
+						event.getMember().getSocketAddress(EndpointQualifier.MEMBER), getCacheListeningPolicy());
+			}
+
+			if (logger.isDebugEnabled()) {
+				logger.debug("Received ItemEvent : " + event);
+			}
+		}
+
+		@Override
+		protected org.springframework.messaging.Message<?> toMessage(ItemEvent<E> event) {
+			final Map<String, Object> headers = new HashMap<>();
+			headers.put(HazelcastHeaders.EVENT_TYPE, event.getEventType().name());
+			headers.put(HazelcastHeaders.MEMBER, event.getMember().getSocketAddress(EndpointQualifier.MEMBER));
+
+			return getMessageBuilderFactory().withPayload(event.getItem()).copyHeaders(headers).build();
+		}
+
+	}
+
+	private class HazelcastMessageListener<E> extends AbstractHazelcastEventListener<Message<E>>
+			implements MessageListener<E> {
+
+		@Override
+		public void onMessage(Message<E> message) {
+			processEvent(message);
+		}
+
+		@Override
+		protected void processEvent(Message<E> event) {
+			sendMessage(event,
+					event.getPublishingMember().getSocketAddress(EndpointQualifier.MEMBER), getCacheListeningPolicy());
+
+			if (logger.isDebugEnabled()) {
+				logger.debug("Received Message : " + event);
+			}
+		}
+
+		@Override
+		protected org.springframework.messaging.Message<?> toMessage(Message<E> event) {
+			Assert.notNull(event.getMessageObject(), "message must not be null");
+
+			final Map<String, Object> headers = new HashMap<>();
+			headers.put(HazelcastHeaders.MEMBER,
+					event.getPublishingMember().getSocketAddress(EndpointQualifier.MEMBER));
+			headers.put(HazelcastHeaders.CACHE_NAME, event.getSource());
+			headers.put(HazelcastHeaders.PUBLISHING_TIME, event.getPublishTime());
+
+			return getMessageBuilderFactory().withPayload(event.getMessageObject()).copyHeaders(headers).build();
+		}
+
+	}
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/inbound/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides classes supporting inbound endpoints.
+ */
+package org.springframework.integration.hazelcast.inbound;

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

@@ -0,0 +1,429 @@
+/*
+ * 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.leader;
+
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.ApplicationEventPublisherAware;
+import org.springframework.context.SmartLifecycle;
+import org.springframework.integration.leader.Candidate;
+import org.springframework.integration.leader.Context;
+import org.springframework.integration.leader.DefaultCandidate;
+import org.springframework.integration.leader.event.DefaultLeaderEventPublisher;
+import org.springframework.integration.leader.event.LeaderEventPublisher;
+import org.springframework.integration.support.leader.LockRegistryLeaderInitiator;
+import org.springframework.util.Assert;
+
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.cp.CPSubsystem;
+import com.hazelcast.cp.lock.FencedLock;
+
+/**
+ * Bootstrap leadership {@link org.springframework.integration.leader.Candidate candidates}
+ * with Hazelcast. Upon construction, {@link #start} must be invoked to
+ * register the candidate for leadership election.
+ *
+ * @author Patrick Peralta
+ * @author Gary Russell
+ * @author Dave Syer
+ * @author Artem Bilan
+ * @author Mael Le Guével
+ * @author Alexey Tsoy
+ * @author Robert Höglund
+ */
+public class LeaderInitiator implements SmartLifecycle, DisposableBean, ApplicationEventPublisherAware {
+
+	private static final Log logger = LogFactory.getLog(LeaderInitiator.class);
+
+	private static int threadNameCount = 0;
+
+	private static final Context NULL_CONTEXT = new NullContext();
+
+	/*** Hazelcast client.
+	 */
+	private final HazelcastInstance client;
+
+	/**
+	 * Candidate for leader election.
+	 */
+	private final Candidate candidate;
+
+	/**
+	 * Executor service for running leadership daemon.
+	 */
+	private final ExecutorService executorService =
+			Executors.newSingleThreadExecutor(r -> {
+				Thread thread = new Thread(r, "Hazelcast-leadership-" + (threadNameCount++));
+				thread.setDaemon(true);
+				return thread;
+			});
+
+	private long heartBeatMillis = LockRegistryLeaderInitiator.DEFAULT_HEART_BEAT_TIME;
+
+	private long busyWaitMillis = LockRegistryLeaderInitiator.DEFAULT_BUSY_WAIT_TIME;
+
+	private LeaderSelector leaderSelector;
+
+	/**
+	 * Leader event publisher.
+	 */
+	private LeaderEventPublisher leaderEventPublisher = new DefaultLeaderEventPublisher();
+
+	private boolean autoStartup = true;
+
+	private int phase;
+
+	/**
+	 * Future returned by submitting an {@link LeaderSelector} to {@link #executorService}.
+	 * This is used to cancel leadership.
+	 */
+	private volatile Future<Void> future;
+
+	private boolean customPublisher = false;
+
+	private volatile boolean running;
+
+	private final Semaphore yieldSign = new Semaphore(0);
+
+	/**
+	 * Construct a {@link LeaderInitiator} with a default candidate.
+	 * @param client Hazelcast client
+	 */
+	public LeaderInitiator(HazelcastInstance client) {
+		this(client, new DefaultCandidate());
+	}
+
+	/**
+	 * Construct a {@link LeaderInitiator}.
+	 * @param client Hazelcast client
+	 * @param candidate leadership election candidate
+	 */
+	public LeaderInitiator(HazelcastInstance client, Candidate candidate) {
+		Assert.notNull(client, "'client' must not be null");
+		Assert.notNull(candidate, "'candidate' must not be null");
+		this.client = client;
+		this.candidate = candidate;
+	}
+
+	/**
+	 * Sets the {@link LeaderEventPublisher}.
+	 * @param leaderEventPublisher the event publisher
+	 */
+	public void setLeaderEventPublisher(LeaderEventPublisher leaderEventPublisher) {
+		Assert.notNull(leaderEventPublisher, "'leaderEventPublisher' must not be null");
+		this.leaderEventPublisher = leaderEventPublisher;
+		this.customPublisher = true;
+	}
+
+	/**
+	 * Time in milliseconds to wait in between attempts to re-acquire the lock, once it is
+	 * held. The heartbeat time has to be less than the remote lock expiry period, if
+	 * there is one, otherwise other nodes can steal the lock while we are sleeping here.
+	 * @param heartBeatMillis the heart-beat timeout in milliseconds.
+	 * Defaults to {@link LockRegistryLeaderInitiator#DEFAULT_HEART_BEAT_TIME}
+	 * @since 1.0.1
+	 */
+	public void setHeartBeatMillis(long heartBeatMillis) {
+		this.heartBeatMillis = heartBeatMillis;
+	}
+
+	/**
+	 * Time in milliseconds to wait in between attempts to acquire the lock, if it is not
+	 * held. The longer this is, the longer the system can be leaderless, if the leader
+	 * dies. If a leader dies without releasing its lock, the system might still have to
+	 * wait for the old lock to expire, but after that it should not have to wait longer
+	 * than the busy wait time to get a new leader.
+	 * @param busyWaitMillis the busy-wait timeout in milliseconds
+	 * Defaults to {@link LockRegistryLeaderInitiator#DEFAULT_BUSY_WAIT_TIME}
+	 * @since 1.0.1
+	 */
+	public void setBusyWaitMillis(long busyWaitMillis) {
+		this.busyWaitMillis = busyWaitMillis;
+	}
+
+	@Override
+	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
+		if (!this.customPublisher) {
+			this.leaderEventPublisher = new DefaultLeaderEventPublisher(applicationEventPublisher);
+		}
+	}
+
+	public void setAutoStartup(boolean autoStartup) {
+		this.autoStartup = autoStartup;
+	}
+
+	@Override
+	public boolean isAutoStartup() {
+		return this.autoStartup;
+	}
+
+	public void setPhase(int phase) {
+		this.phase = phase;
+	}
+
+	@Override
+	public int getPhase() {
+		return this.phase;
+	}
+
+	/**
+	 * The context of the initiator or null if not running.
+	 * @return the context (or null if not running)
+	 */
+	public Context getContext() {
+		if (this.leaderSelector == null) {
+			return NULL_CONTEXT;
+		}
+		return this.leaderSelector.context;
+	}
+
+	/**
+	 * Start the registration of the {@link #candidate} for leader election.
+	 */
+	@Override
+	public synchronized void start() {
+		if (!this.running) {
+			this.leaderSelector = new LeaderSelector();
+			this.running = true;
+			this.future = this.executorService.submit(this.leaderSelector);
+		}
+	}
+
+	@Override
+	public void stop(Runnable callback) {
+		stop();
+		callback.run();
+	}
+
+	/**
+	 * Stop the registration of the {@link #candidate} for leader election.
+	 * If the candidate is currently leader, its leadership will be revoked.
+	 */
+	@Override
+	public synchronized void stop() {
+		if (this.running) {
+			this.running = false;
+			if (this.future != null) {
+				this.future.cancel(true);
+			}
+			this.future = null;
+		}
+	}
+
+	/**
+	 * {@code true} if leadership election for this {@link #candidate} is running.
+	 * @return true if leadership election for this {@link #candidate} is running
+	 */
+	@Override
+	public boolean isRunning() {
+		return this.running;
+	}
+
+	@Override
+	public void destroy() {
+		stop();
+		this.executorService.shutdown();
+	}
+
+
+	FencedLock getLock() {
+		CPSubsystem cpSubSystem = this.client.getCPSubsystem();
+		FencedLock lock = cpSubSystem.getLock(this.candidate.getRole());
+		if (logger.isDebugEnabled()) {
+			logger.debug(
+					String.format("Use lock groupId '%s', lock count '%s'", lock.getGroupId(), lock.getLockCount()));
+		}
+		return lock;
+	}
+
+
+	/**
+	 * Callable that manages the acquisition of Hazelcast locks
+	 * for leadership election.
+	 */
+	protected class LeaderSelector implements Callable<Void> {
+
+		protected final HazelcastContext context = new HazelcastContext();
+
+		protected final String role = LeaderInitiator.this.candidate.getRole();
+
+		private volatile boolean leader = false;
+
+		@Override
+		public Void call() {
+			try {
+				while (isRunning()) {
+					try {
+						if (logger.isTraceEnabled()) {
+							logger.trace("Am I the leader (" + LeaderInitiator.this.candidate.getRole() + ") ? "
+									+ this.leader);
+						}
+						if (getLock().isLockedByCurrentThread()) {
+							if (!this.leader) {
+								// Since we have the lock we need to ensure that the leader flag is set
+								this.leader = true;
+							}
+							// Give it a chance to expire.
+							if (LeaderInitiator.this.yieldSign.tryAcquire(LeaderInitiator.this.heartBeatMillis, TimeUnit.MILLISECONDS)) {
+								revokeLeadership();
+								// Give it a chance to elect some other leader.
+								Thread.sleep(LeaderInitiator.this.busyWaitMillis);
+							}
+						}
+						else {
+							// We try to acquire the lock
+							boolean acquired = getLock()
+									.tryLock(LeaderInitiator.this.heartBeatMillis, TimeUnit.MILLISECONDS);
+							if (acquired && !this.leader) {
+								// Success: we are now leader
+								this.leader = true;
+								handleGranted();
+							}
+						}
+					}
+					catch (Exception e) {
+						// The lock was broken and we are no longer leader
+						revokeLeadership();
+
+
+						if (isRunning()) {
+							// Give it a chance to elect some other leader.
+							try {
+								Thread.sleep(LeaderInitiator.this.busyWaitMillis);
+							}
+							catch (InterruptedException e1) {
+								// Ignore interruption and let it to be caught on the next cycle.
+								Thread.currentThread().interrupt();
+							}
+						}
+						if (logger.isDebugEnabled()) {
+							logger.debug("Error acquiring the lock for " + this.context +
+									". " + (isRunning() ? "Retrying..." : ""), e);
+						}
+					}
+				}
+			}
+			finally {
+				revokeLeadership();
+			}
+
+			return null;
+		}
+
+		private void revokeLeadership() {
+			if (this.leader) {
+				this.leader = false;
+				try {
+					// Try to unlock
+					getLock().unlock();
+				}
+				catch (Exception e1) {
+					logger.warn("Could not unlock - treat as broken " + this.context + ". Revoking "
+							+ (isRunning() ? " and retrying..." : "..."), e1);
+
+				}
+
+				handleRevoked();
+			}
+		}
+
+
+		private void handleGranted() throws InterruptedException {
+			LeaderInitiator.this.candidate.onGranted(this.context);
+			if (LeaderInitiator.this.leaderEventPublisher != null) {
+				try {
+					LeaderInitiator.this.leaderEventPublisher.publishOnGranted(
+							LeaderInitiator.this, this.context, this.role);
+				}
+				catch (Exception e) {
+					logger.warn("Error publishing OnGranted event.", e);
+				}
+			}
+		}
+
+		private void handleRevoked() {
+			LeaderInitiator.this.candidate.onRevoked(this.context);
+			if (LeaderInitiator.this.leaderEventPublisher != null) {
+				try {
+					LeaderInitiator.this.leaderEventPublisher.publishOnRevoked(
+							LeaderInitiator.this, this.context, this.role);
+				}
+				catch (Exception e) {
+					logger.warn("Error publishing OnRevoked event.", e);
+				}
+			}
+		}
+
+	}
+
+	/**
+	 * Implementation of leadership context backed by Hazelcast.
+	 */
+	protected class HazelcastContext implements Context {
+
+		@Override
+		public boolean isLeader() {
+			return LeaderInitiator.this.leaderSelector.leader;
+		}
+
+		@Override
+		public void yield() {
+			if (isLeader()) {
+				LeaderInitiator.this.yieldSign.release();
+			}
+		}
+
+		@Override
+		public String getRole() {
+			return LeaderInitiator.this.candidate.getRole();
+		}
+
+		@Override
+		public String toString() {
+			return "HazelcastContext{role=" + LeaderInitiator.this.candidate.getRole() +
+					", id=" + LeaderInitiator.this.candidate.getId() +
+					", isLeader=" + isLeader() + "}";
+		}
+
+	}
+
+	private static final class NullContext implements Context {
+
+		@Override
+		public boolean isLeader() {
+			return false;
+		}
+
+		@Override
+		public void yield() {
+			// No-op
+		}
+
+	}
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/leader/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides the Leader Initiator support classes.
+ */
+package org.springframework.integration.hazelcast.leader;

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

@@ -0,0 +1,88 @@
+/*
+ * Copyright 2015-2019 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.listener;
+
+import java.net.SocketAddress;
+import java.util.Set;
+import java.util.concurrent.locks.Lock;
+
+import org.springframework.integration.hazelcast.HazelcastLocalInstanceRegistrar;
+
+import com.hazelcast.cluster.MembershipAdapter;
+import com.hazelcast.cluster.MembershipEvent;
+import com.hazelcast.core.Hazelcast;
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.instance.EndpointQualifier;
+import com.hazelcast.multimap.MultiMap;
+
+/**
+ * Hazelcast {@link MembershipAdapter} in order to listen for membership updates in the cluster.
+ *
+ * @author Eren Avsarogullari
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+public class HazelcastMembershipListener extends MembershipAdapter {
+
+	@Override
+	public void memberRemoved(MembershipEvent membershipEvent) {
+		SocketAddress removedMemberSocketAddress =
+				membershipEvent.getMember().getSocketAddress(EndpointQualifier.MEMBER);
+		Set<HazelcastInstance> hazelcastLocalInstanceSet = Hazelcast.getAllHazelcastInstances();
+		if (!hazelcastLocalInstanceSet.isEmpty()) {
+			HazelcastInstance hazelcastInstance = hazelcastLocalInstanceSet.iterator().next();
+			Lock lock =
+					hazelcastInstance.getCPSubsystem()
+							.getLock(HazelcastLocalInstanceRegistrar.SPRING_INTEGRATION_INTERNAL_CLUSTER_LOCK);
+			lock.lock();
+			try {
+				MultiMap<SocketAddress, SocketAddress> configMultiMap = hazelcastInstance
+						.getMultiMap(HazelcastLocalInstanceRegistrar.SPRING_INTEGRATION_INTERNAL_CLUSTER_MULTIMAP);
+
+				if (configMultiMap.containsKey(removedMemberSocketAddress)) {
+					SocketAddress newAdminSocketAddress = getNewAdminInstanceSocketAddress(
+							configMultiMap, removedMemberSocketAddress);
+					for (SocketAddress socketAddress : configMultiMap.values()) {
+						if (!socketAddress.equals(removedMemberSocketAddress)) {
+							configMultiMap.put(newAdminSocketAddress, socketAddress);
+						}
+					}
+					configMultiMap.remove(removedMemberSocketAddress);
+				}
+				else {
+					configMultiMap.remove(configMultiMap.keySet().iterator().next(), removedMemberSocketAddress);
+				}
+			}
+			finally {
+				lock.unlock();
+			}
+		}
+	}
+
+	private SocketAddress getNewAdminInstanceSocketAddress(
+			MultiMap<SocketAddress, SocketAddress> configMultiMap, SocketAddress removedMemberSocketAddress) {
+		for (SocketAddress socketAddress : configMultiMap.values()) {
+			if (!socketAddress.equals(removedMemberSocketAddress)) {
+				return socketAddress;
+			}
+		}
+
+		throw new IllegalStateException("No Active Hazelcast Instance Found.");
+	}
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/listener/package-info.java

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

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

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2017-2019 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.locks.Lock;
+
+import org.springframework.integration.support.locks.LockRegistry;
+import org.springframework.util.Assert;
+
+import com.hazelcast.core.HazelcastInstance;
+
+/**
+ * A {@link LockRegistry} implementation Hazelcast distributed locks.
+ *
+ * @author Artem Bilan
+ */
+public class HazelcastLockRegistry implements LockRegistry {
+
+	private final HazelcastInstance client;
+
+	public HazelcastLockRegistry(HazelcastInstance hazelcastInstance) {
+		Assert.notNull(hazelcastInstance, "'hazelcastInstance' must not be null");
+		this.client = hazelcastInstance;
+	}
+
+	@Override
+	public Lock obtain(Object lockKey) {
+		Assert.isInstanceOf(String.class, lockKey);
+		return this.client.getCPSubsystem().getLock((String) lockKey);
+	}
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/lock/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides the distributed Locks support classes.
+ */
+package org.springframework.integration.hazelcast.lock;

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/message/EntryEventMessagePayload.java

@@ -0,0 +1,86 @@
+/*
+ * 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.message;
+
+import org.springframework.util.Assert;
+
+/**
+ * Hazelcast Message Payload for Entry Events.
+ *
+ * @param <K> the entry key type
+ * @param <V> the entry value type
+ *
+ * @author Eren Avsarogullari
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+public class EntryEventMessagePayload<K, V> {
+
+	/**
+	 * The entry key.
+	 */
+	public final K key;
+
+	/**
+	 * The entry value.
+	 */
+	public final V value;
+
+	/**
+	 * The entry old value if any.
+	 */
+	public final V oldValue;
+
+	public EntryEventMessagePayload(final K key, final V value, final V oldValue) {
+		Assert.notNull(key, "'key' must not be null");
+		this.key = key;
+		this.value = value;
+		this.oldValue = oldValue;
+	}
+
+	@Override
+	public String toString() {
+		return "EntryEventMessagePayload [key=" + this.key + ", value=" + this.value + ", oldValue=" + this.oldValue + "]";
+	}
+
+	@Override
+	public boolean equals(Object o) {
+		if (this == o) {
+			return true;
+		}
+		if (o == null || getClass() != o.getClass()) {
+			return false;
+		}
+
+		EntryEventMessagePayload<?, ?> that = (EntryEventMessagePayload<?, ?>) o;
+
+		return this.key.equals(that.key) && !(this.value != null ? !this.value.equals(that.value)
+				: that.value != null) && !(this.oldValue != null
+				? !this.oldValue.equals(that.oldValue) : that.oldValue != null);
+
+	}
+
+	@Override
+	public int hashCode() {
+		int result = this.key.hashCode();
+		result = 31 * result + (this.value != null ? this.value.hashCode() : 0);
+		result = 31 * result + (this.oldValue != null ? this.oldValue.hashCode() : 0);
+		return result;
+	}
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/message/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides classes supporting Hazelcast message headers and payload.
+ */
+package org.springframework.integration.hazelcast.message;

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/metadata/HazelcastMetadataStore.java

@@ -0,0 +1,140 @@
+/*
+ * Copyright 2017 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.metadata;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.integration.metadata.ListenableMetadataStore;
+import org.springframework.integration.metadata.MetadataStoreListener;
+import org.springframework.util.Assert;
+
+import com.hazelcast.core.EntryEvent;
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.map.IMap;
+import com.hazelcast.map.listener.EntryAddedListener;
+import com.hazelcast.map.listener.EntryRemovedListener;
+import com.hazelcast.map.listener.EntryUpdatedListener;
+
+/**
+ * The Hazelcast {@link IMap}-based {@link ListenableMetadataStore} implementation.
+ *
+ * @author Vinicius Carvalho
+ * @author Artem Bilan
+ */
+public class HazelcastMetadataStore implements ListenableMetadataStore, InitializingBean {
+
+	private static final String METADATA_STORE_MAP_NAME = "SPRING_INTEGRATION_METADATA_STORE";
+
+	private final IMap<String, String> map;
+
+	private final List<MetadataStoreListener> listeners = new CopyOnWriteArrayList<MetadataStoreListener>();
+
+	public HazelcastMetadataStore(HazelcastInstance hazelcastInstance) {
+		Assert.notNull(hazelcastInstance, "Hazelcast instance can't be null");
+		this.map = hazelcastInstance.getMap(METADATA_STORE_MAP_NAME);
+	}
+
+	public HazelcastMetadataStore(IMap<String, String> map) {
+		Assert.notNull(map, "IMap reference can not be null");
+		this.map = map;
+	}
+
+	@Override
+	public String putIfAbsent(String key, String value) {
+		Assert.notNull(key, "'key' must not be null.");
+		Assert.notNull(value, "'value' must not be null.");
+		return this.map.putIfAbsent(key, value);
+	}
+
+	@Override
+	public boolean replace(String key, String oldValue, String newValue) {
+		Assert.notNull(key, "'key' must not be null.");
+		Assert.notNull(oldValue, "'oldValue' must not be null.");
+		Assert.notNull(newValue, "'newValue' must not be null.");
+		return this.map.replace(key, oldValue, newValue);
+	}
+
+	@Override
+	public void put(String key, String value) {
+		Assert.notNull(key, "'key' must not be null.");
+		Assert.notNull(value, "'value' must not be null.");
+		this.map.put(key, value);
+	}
+
+	@Override
+	public String get(String key) {
+		Assert.notNull(key, "'key' must not be null.");
+		return this.map.get(key);
+	}
+
+	@Override
+	public String remove(String key) {
+		Assert.notNull(key, "'key' must not be null.");
+		return this.map.remove(key);
+	}
+
+	@Override
+	public void addListener(MetadataStoreListener callback) {
+		Assert.notNull(callback, "callback object can not be null");
+		this.listeners.add(callback);
+	}
+
+	@Override
+	public void removeListener(MetadataStoreListener callback) {
+		this.listeners.remove(callback);
+	}
+
+	@Override
+	public void afterPropertiesSet() throws Exception {
+		this.map.addEntryListener(new MapListener(this.listeners), true);
+	}
+
+	private static class MapListener implements EntryAddedListener<String, String>,
+			EntryRemovedListener<String, String>, EntryUpdatedListener<String, String> {
+
+		private final List<MetadataStoreListener> listeners;
+
+		MapListener(List<MetadataStoreListener> listeners) {
+			this.listeners = listeners;
+		}
+
+		@Override
+		public void entryAdded(EntryEvent<String, String> event) {
+			for (MetadataStoreListener listener : this.listeners) {
+				listener.onAdd(event.getKey(), event.getValue());
+			}
+		}
+
+		@Override
+		public void entryRemoved(EntryEvent<String, String> event) {
+			for (MetadataStoreListener listener : this.listeners) {
+				listener.onRemove(event.getKey(), event.getOldValue());
+			}
+		}
+
+		@Override
+		public void entryUpdated(EntryEvent<String, String> event) {
+			for (MetadataStoreListener listener : this.listeners) {
+				listener.onUpdate(event.getKey(), event.getValue());
+			}
+		}
+
+	}
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/metadata/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides the Metadata Store support classes.
+ */
+package org.springframework.integration.hazelcast.metadata;

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/outbound/HazelcastCacheWritingMessageHandler.java

@@ -0,0 +1,167 @@
+/*
+ * Copyright 2015-2019 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.outbound;
+
+import java.util.Collection;
+import java.util.Map;
+
+import org.springframework.expression.EvaluationContext;
+import org.springframework.expression.Expression;
+import org.springframework.integration.expression.ExpressionUtils;
+import org.springframework.integration.handler.AbstractMessageHandler;
+import org.springframework.integration.hazelcast.HazelcastHeaders;
+import org.springframework.messaging.Message;
+import org.springframework.util.Assert;
+
+import com.hazelcast.core.DistributedObject;
+import com.hazelcast.multimap.MultiMap;
+import com.hazelcast.topic.ITopic;
+
+/**
+ * MessageHandler implementation that writes {@link Message} or payload to defined
+ * Hazelcast distributed cache object.
+ *
+ * @author Eren Avsarogullari
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+public class HazelcastCacheWritingMessageHandler extends AbstractMessageHandler {
+
+	private DistributedObject distributedObject;
+
+	private Expression cacheExpression;
+
+	private Expression keyExpression;
+
+	private boolean extractPayload = true;
+
+	private EvaluationContext evaluationContext;
+
+	public void setDistributedObject(DistributedObject distributedObject) {
+		Assert.notNull(distributedObject, "'distributedObject' must not be null");
+		this.distributedObject = distributedObject;
+	}
+
+	public void setCacheExpression(Expression cacheExpression) {
+		Assert.notNull(cacheExpression, "'cacheExpression' must not be null");
+		this.cacheExpression = cacheExpression;
+	}
+
+	public void setKeyExpression(Expression keyExpression) {
+		Assert.notNull(keyExpression, "'keyExpression' must not be null");
+		this.keyExpression = keyExpression;
+	}
+
+	public void setExtractPayload(boolean extractPayload) {
+		this.extractPayload = extractPayload;
+	}
+
+	@Override
+	protected void onInit() {
+		super.onInit();
+		this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
+	}
+
+	@Override
+	@SuppressWarnings({"unchecked", "rawtypes"})
+	protected void handleMessageInternal(final Message<?> message) {
+		Object objectToStore = message;
+		if (this.extractPayload) {
+			objectToStore = message.getPayload();
+		}
+
+		DistributedObject distributedObject = getDistributedObject(message);
+
+		if (distributedObject instanceof Map) {
+			Map map = (Map) distributedObject;
+			if (objectToStore instanceof Map) {
+				map.putAll((Map) objectToStore);
+			}
+			else if (objectToStore instanceof Map.Entry) {
+				Map.Entry entry = (Map.Entry) objectToStore;
+				map.put(entry.getKey(), entry.getValue());
+			}
+			else {
+				map.put(getKey(message), objectToStore);
+			}
+		}
+		else if (distributedObject instanceof MultiMap) {
+			MultiMap map = (MultiMap) distributedObject;
+			if (objectToStore instanceof Map) {
+				Map<?, ?> mapToStore = (Map) objectToStore;
+				for (Map.Entry entry : mapToStore.entrySet()) {
+					map.put(entry.getKey(), entry.getValue());
+				}
+			}
+			else if (objectToStore instanceof Map.Entry) {
+				Map.Entry entry = (Map.Entry) objectToStore;
+				map.put(entry.getKey(), entry.getValue());
+			}
+			else {
+				map.put(getKey(message), objectToStore);
+			}
+		}
+		else if (distributedObject instanceof ITopic) {
+			((ITopic) distributedObject).publish(objectToStore);
+		}
+		else if (distributedObject instanceof Collection) {
+			if (objectToStore instanceof Collection) {
+				((Collection) distributedObject).addAll((Collection) objectToStore);
+			}
+			else {
+				((Collection) distributedObject).add(objectToStore);
+			}
+		}
+		else {
+			throw new IllegalStateException("The 'distributedObject' for 'HazelcastCacheWritingMessageHandler' " +
+					"must be of 'IMap', 'MultiMap', 'ITopic', 'ISet' or 'IList' type, " +
+					"but gotten: [" + distributedObject + "].");
+		}
+	}
+
+
+	private DistributedObject getDistributedObject(final Message<?> message) {
+		if (this.distributedObject != null) {
+			return this.distributedObject;
+		}
+		else if (this.cacheExpression != null) {
+			return this.cacheExpression.getValue(this.evaluationContext, message, DistributedObject.class);
+		}
+		else if (message.getHeaders().containsKey(HazelcastHeaders.CACHE_NAME)) {
+			return getBeanFactory()
+					.getBean(message.getHeaders().get(HazelcastHeaders.CACHE_NAME, String.class),
+							DistributedObject.class);
+		}
+		else {
+			throw new IllegalStateException("One of 'cache', 'cache-expression' and "
+					+ HazelcastHeaders.CACHE_NAME
+					+ " must be set for cache object definition.");
+		}
+	}
+
+	private Object getKey(Message<?> message) {
+		if (this.keyExpression != null) {
+			return this.keyExpression.getValue(this.evaluationContext, message);
+		}
+		else {
+			throw new IllegalStateException(
+					"'key-expression' must be set to place the raw 'payload' to the IMap, MultiMap and ReplicatedMap");
+		}
+	}
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/outbound/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides classes supporting outbound endpoints.
+ */
+package org.springframework.integration.hazelcast.outbound;

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides common used types and classes.
+ */
+package org.springframework.integration.hazelcast;

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/store/HazelcastMessageStore.java

@@ -0,0 +1,83 @@
+/*
+ * Copyright 2017-2021 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.store;
+
+import java.util.Collection;
+
+import org.springframework.integration.store.AbstractKeyValueMessageStore;
+import org.springframework.util.Assert;
+
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.map.IMap;
+import com.hazelcast.query.Predicates;
+import com.hazelcast.query.QueryConstants;
+
+/**
+ * The Hazelcast {@link IMap}-based {@link AbstractKeyValueMessageStore} implementation.
+ *
+ * @author Vinicius Carvalho
+ * @author Artem Bilan
+ */
+public class HazelcastMessageStore extends AbstractKeyValueMessageStore {
+
+	private static final String MESSAGE_STORE_MAP_NAME = "SPRING_INTEGRATION_MESSAGE_STORE";
+
+	private final IMap<Object, Object> map;
+
+	public HazelcastMessageStore(HazelcastInstance hazelcastInstance) {
+		Assert.notNull(hazelcastInstance, "Hazelcast instance can't be null");
+		this.map = hazelcastInstance.getMap(MESSAGE_STORE_MAP_NAME);
+	}
+
+	public HazelcastMessageStore(IMap<Object, Object> map) {
+		Assert.notNull(map, "IMap reference can not be null");
+		this.map = map;
+	}
+
+	@Override
+	protected Object doRetrieve(Object id) {
+		return this.map.get(id);
+	}
+
+	@Override
+	protected void doStore(Object id, Object objectToStore) {
+		this.map.put(id, objectToStore);
+	}
+
+	@Override
+	protected void doStoreIfAbsent(Object id, Object objectToStore) {
+		this.map.putIfAbsent(id, objectToStore);
+	}
+
+	@Override
+	protected void doRemoveAll(Collection<Object> ids) {
+		this.map.removeAll((mapEntry) -> ids.contains(mapEntry.getKey()));
+	}
+
+	@Override
+	protected Object doRemove(Object id) {
+		return this.map.remove(id);
+	}
+
+	@Override
+	protected Collection<?> doListKeys(String keyPattern) {
+		Assert.hasText(keyPattern, "'keyPattern' must not be empty");
+		keyPattern = keyPattern.replaceAll("\\*", "%");
+		return this.map.keySet(Predicates.like(QueryConstants.KEY_ATTRIBUTE_NAME.value(), keyPattern));
+	}
+
+}

spring-integration-hazelcast/src/main/java/org/springframework/integration/hazelcast/store/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides the Message Store support classes.
+ */
+package org.springframework.integration.hazelcast.store;

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

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

spring-integration-hazelcast/src/main/resources/META-INF/spring.handlers

@@ -0,0 +1 @@
+http\://www.springframework.org/schema/integration/hazelcast=org.springframework.integration.hazelcast.config.xml.HazelcastIntegrationNamespaceHandler

spring-integration-hazelcast/src/main/resources/META-INF/spring.schemas

@@ -0,0 +1,4 @@
+http\://www.springframework.org/schema/integration/hazelcast/spring-integration-hazelcast-1.0.xsd=org/springframework/integration/hazelcast/config/xml/spring-integration-hazelcast.xsd
+http\://www.springframework.org/schema/integration/hazelcast/spring-integration-hazelcast.xsd=org/springframework/integration/hazelcast/config/xml/spring-integration-hazelcast.xsd
+https\://www.springframework.org/schema/integration/hazelcast/spring-integration-hazelcast-1.0.xsd=org/springframework/integration/hazelcast/config/xml/spring-integration-hazelcast.xsd
+https\://www.springframework.org/schema/integration/hazelcast/spring-integration-hazelcast.xsd=org/springframework/integration/hazelcast/config/xml/spring-integration-hazelcast.xsd

spring-integration-hazelcast/src/main/resources/org/springframework/integration/hazelcast/config/xml/spring-integration-hazelcast.xsd

@@ -0,0 +1,303 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<xsd:schema xmlns="http://www.springframework.org/schema/integration/hazelcast"
+	xmlns:xsd="http://www.w3.org/2001/XMLSchema"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xmlns:tool="http://www.springframework.org/schema/tool"
+	xmlns:integration="http://www.springframework.org/schema/integration"
+	targetNamespace="http://www.springframework.org/schema/integration/hazelcast"
+	elementFormDefault="qualified"
+	attributeFormDefault="unqualified">
+
+	<xsd:import namespace="http://www.springframework.org/schema/beans"
+				schemaLocation="https://www.springframework.org/schema/beans/spring-beans.xsd"/>
+	<xsd:import namespace="http://www.springframework.org/schema/tool"/>
+	<xsd:import namespace="http://www.springframework.org/schema/integration"
+				schemaLocation="https://www.springframework.org/schema/integration/spring-integration.xsd"/>
+
+	<xsd:element name="inbound-channel-adapter">
+
+		<xsd:annotation>
+			<xsd:documentation>
+				Configures Hazelcast Event-Driven Inbound Channel Adapter
+			</xsd:documentation>
+		</xsd:annotation>
+
+		<xsd:complexType>
+			<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
+
+			<xsd:attribute name="cache" use="required" type="xsd:string">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="com.hazelcast.core.DistributedObject" />
+						</tool:annotation>
+					</xsd:appinfo>
+					<xsd:documentation>
+						<![CDATA[ Specifies cache reference to listen ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+
+			<xsd:attribute name="cache-events" type="xsd:string" use="optional" default="ADDED">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="value">
+							<tool:expected-type
+								type="org.springframework.integration.hazelcast.common.CacheEventType" />
+						</tool:annotation>
+					</xsd:appinfo>
+					<xsd:documentation>
+						<![CDATA[ Specifies cache entry event types ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+
+			<xsd:attribute name="cache-listening-policy" default="SINGLE" use="optional">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[ Specifies cache listening policy. ]]></xsd:documentation>
+				</xsd:annotation>
+				<xsd:simpleType>
+					<xsd:restriction base="xsd:string">
+						<xsd:enumeration value="SINGLE" />
+						<xsd:enumeration value="ALL" />
+					</xsd:restriction>
+				</xsd:simpleType>
+			</xsd:attribute>
+
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:element name="outbound-channel-adapter">
+
+		<xsd:annotation>
+			<xsd:documentation>
+				Configures Hazelcast Outbound Channel Adapter
+			</xsd:documentation>
+		</xsd:annotation>
+
+		<xsd:complexType>
+			<xsd:choice minOccurs="0" maxOccurs="2">
+				<xsd:element name="request-handler-advice-chain" type="integration:handlerAdviceChainType"
+							minOccurs="0" maxOccurs="1" />
+				<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
+			</xsd:choice>
+
+ 			<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
+
+			<xsd:attribute name="cache" type="xsd:string" use="optional">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="com.hazelcast.core.DistributedObject" />
+						</tool:annotation>
+					</xsd:appinfo>
+					<xsd:documentation>
+						<![CDATA[ Specifies cache reference to listen ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			
+			<xsd:attribute name="cache-expression" type="xsd:string" use="optional">
+				<xsd:annotation>
+					<xsd:documentation>
+						<![CDATA[ Specifies cache name to listen ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			
+			<xsd:attribute name="key-expression" type="xsd:string" use="optional">
+				<xsd:annotation>
+					<xsd:documentation>
+						<![CDATA[ Specifies entry key ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			
+			<xsd:attribute name="extract-payload" type="xsd:boolean" default="true">
+				<xsd:annotation>
+					<xsd:documentation>
+						<![CDATA[ Specifies whole message or just payload to send ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+
+			<xsd:attribute name="order" type="xsd:string" use="optional">
+                <xsd:annotation>
+                    <xsd:documentation><![CDATA[
+                    Specifies the order for invocation when this endpoint is connected as a
+                    subscriber to a SubscribableChannel.
+                ]]></xsd:documentation>
+                </xsd:annotation>
+            </xsd:attribute>
+
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:element name="cq-inbound-channel-adapter">
+
+		<xsd:annotation>
+			<xsd:documentation>
+				Configures Hazelcast Continuous Query Inbound Channel Adapter
+			</xsd:documentation>
+		</xsd:annotation>
+
+		<xsd:complexType>
+
+			<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
+
+			<xsd:attribute name="cache" use="required" type="xsd:string">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="com.hazelcast.core.IMap" />
+						</tool:annotation>
+					</xsd:appinfo>
+					<xsd:documentation>
+						<![CDATA[ Specifies cache reference to listen ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+
+			<xsd:attribute name="cache-events" type="xsd:string" use="optional" default="ADDED">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="value">
+							<tool:expected-type
+								type="org.springframework.integration.hazelcast.common.CacheEventType" />
+						</tool:annotation>
+					</xsd:appinfo>
+					<xsd:documentation>
+						<![CDATA[ Specifies cache entry event types ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+
+			<xsd:attribute name="predicate" type="xsd:string" use="required">
+				<xsd:annotation>
+					<xsd:documentation>
+						<![CDATA[ Specifies predicate for continuous query ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+
+			<xsd:attribute name="include-value" type="xsd:boolean" default="true">
+				<xsd:annotation>
+					<xsd:documentation>
+						<![CDATA[ Specifies including of value and oldValue in continuous query result ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+
+			<xsd:attribute name="cache-listening-policy" default="SINGLE" use="optional">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[ Specifies cache listening policy. ]]></xsd:documentation>
+				</xsd:annotation>
+				<xsd:simpleType>
+					<xsd:restriction base="xsd:string">
+						<xsd:enumeration value="SINGLE" />
+						<xsd:enumeration value="ALL" />
+					</xsd:restriction>
+				</xsd:simpleType>
+			</xsd:attribute>
+
+		</xsd:complexType>
+	</xsd:element>
+
+	<xsd:element name="ds-inbound-channel-adapter">
+
+		<xsd:annotation>
+			<xsd:documentation>
+				Configures Hazelcast Distributed SQL Inbound Channel Adapter
+			</xsd:documentation>
+		</xsd:annotation>
+
+		<xsd:complexType>
+
+			<xsd:sequence>
+				<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
+			</xsd:sequence>
+
+			<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
+
+			<xsd:attribute name="cache" use="required" type="xsd:string">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="com.hazelcast.core.IMap" />
+						</tool:annotation>
+					</xsd:appinfo>
+					<xsd:documentation>
+						<![CDATA[ Specifies cache reference to listen ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+
+			<xsd:attribute name="iteration-type" default="VALUE" use="optional">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[ Specifies Distributed-SQL Iteration Types. ]]></xsd:documentation>
+				</xsd:annotation>
+				<xsd:simpleType>
+					<xsd:restriction base="xsd:string">
+						<xsd:enumeration value="ENTRY" />
+						<xsd:enumeration value="KEY" />
+						<xsd:enumeration value="LOCAL_KEY" />
+						<xsd:enumeration value="VALUE" />
+					</xsd:restriction>
+				</xsd:simpleType>
+			</xsd:attribute>
+
+			<xsd:attribute name="distributed-sql" type="xsd:string" use="required">
+				<xsd:annotation>
+					<xsd:documentation>
+						<![CDATA[ Specifies Distributed-SQL ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+
+		</xsd:complexType>
+
+	</xsd:element>
+	
+	<xsd:element name="cm-inbound-channel-adapter">
+
+		<xsd:annotation>
+			<xsd:documentation>
+				Configures Hazelcast Cluster Monitor Inbound Channel Adapter
+			</xsd:documentation>
+		</xsd:annotation>
+
+		<xsd:complexType>
+
+			<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
+
+			<xsd:attribute name="hazelcast-instance" use="required" type="xsd:string">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="com.hazelcast.core.HazelcastInstance" />
+						</tool:annotation>
+					</xsd:appinfo>
+					<xsd:documentation>
+						<![CDATA[ Specifies hazelcast instance reference to listen ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+
+			<xsd:attribute name="monitor-types" type="xsd:string" use="optional" default="MEMBERSHIP">
+				<xsd:annotation>
+					<xsd:appinfo>
+						<tool:annotation kind="value">
+							<tool:expected-type
+								type="org.springframework.integration.hazelcast.ClusterMonitorType" />
+						</tool:annotation>
+					</xsd:appinfo>
+					<xsd:documentation>
+						<![CDATA[ Specifies cluster monitor types ]]>
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+		</xsd:complexType>
+
+	</xsd:element>
+	
+</xsd:schema>