When configuring a HazelcastEventDrivenMessageProducer I'm not seeing the message actually dequeue from my hazelcast queue.

[vsalvati]

I'm trying to configure a mechanism using spring integration to consume from a hazelcast queue into my spring boot project running on Java 11

I'm using spring integration with the following dependencies in my pom.xml

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-integration</artifactId>
</dependency>
<dependency>
  <groupId>org.springframework.integration</groupId>
  <artifactId>spring-integration-hazelcast</artifactId>
  <version>3.0.0</version>
</dependency>

No matter how I configure my HazelcastEventDrivenMessageProducer, MessageChannel, and QueueChannel I can't get my consumer code to run once per message and dequeue from my hazelcast queue. I am trying to read from an externalized queue in my external hazelcast cluster and dequeue the message once done.

Here is my configuration setup..

 @Bean
    HazelcastEventDrivenMessageProducer hazelcastEventDrivenMessageProducer(@Qualifier("myQueue") IQueue<String> myQueue) {
        HazelcastEventDrivenMessageProducer hazelcastEventDrivenMessageProducer = new HazelcastEventDrivenMessageProducer(myQueue);
        hazelcastEventDrivenMessageProducer.setOutputChannel(hazelcastQueueChannel());
        hazelcastEventDrivenMessageProducer.setCacheEventTypes("ADDED,REMOVED");
        hazelcastEventDrivenMessageProducer.setCacheListeningPolicy(CacheListeningPolicyType.SINGLE);
        return hazelcastEventDrivenMessageProducer;
    }
    
    @Bean
    IQueue<String> myQueue(@Qualifier("externalHazelcastInstance") HazelcastInstance hazelcastInstance) {
        return hazelcastInstance.getQueue("myQueue");
    }

    @Bean
    public PollableChannel hazelcastQueueChannel() {
        return new QueueChannel();
    }

I also have a consumer class that looks like this.

@Slf4j
@Component
public class QueueTaskConsumer {
    @ServiceActivator(inputChannel = "hazelcastQueueChannel")
    public void readFromTaskQueue(String message) {
        log.info("received the message inside QueueTaskConsumer " + message);
    }
}

To Reproduce

Configure your setup as the same above and try to add 10 messages(String) to a hazelcast queue

Expected behavior

I expect to see the output of "received the message inside QueueTaskConsumer" in my test 10 times and the queue to be empty.

Currently I have 17 messages in my queue and I have a test that writes 10 messages to my queue with my consumer set up this way. I can see the logging of these 10 messages with "received the message inside QueueTaskConsumer" but the other 17 remain and now I have 27 messages in my queue. It never seems to dequeue at all from my queue.

The consumer should have logged those 10 + the 17 that already existed and dequeued all 27.

I have also tried adding a consumer inside my class representing a Spring Boot app with @SpringBootTest. There I try to inject my channel itself and call receive and the item variable is always null when trying to get from the channel.

@SpringBootApplication(scanBasePackages = "com.mypackage")
public class ExternalHazelcastApp {
    public static void main(String[] args) {
        SpringApplication app = new SpringApplication(ExternalHazelcastApp.class);
        app.setDefaultProperties(Collections.singletonMap("server.port", "8085"));
        app.run(args);
    }

    @Component
    @Slf4j
    public static class QueueTaskConsumerInternal {
        @Autowired @Qualifier("hazelcastQueueChannel") public PollableChannel hazelcastQueueChannel;

        @EventListener(value = ApplicationReadyEvent.class)
        public void start() {
            log.info("CONSUMER HAS STARTED");
            while (true)
            {
                try {
                    log.info("INSIDE WHILE LOOP");
                    Message<?> item = hazelcastQueueChannel.receive(1_000); // I always get a null here and a npe on line below.
                    log.info("Consumed: " + item.getPayload());
                }
                catch (Exception e) {
                    log.info("exception " + e.getCause());
                    e.printStackTrace();
                }
            }
        }
    }

I have seen my queue actually dequeue messages when my queue setup is like the following below. With this though, I see consumption code run 50 times for one message and that cannot happen. The following is how i configured it.

@Bean
    IQueue<Message<?>> myQueue(@Qualifier("externalHazelcastInstance") HazelcastInstance hazelcastInstance) {
        return hazelcastInstance.getQueue(MY_TEST_QUEUE);
    }

    @Bean
    public MessageChannel hazelcastQueueChannel(@Qualifier("myQueue") IQueue<Message<?>> myQueue) {
        return new QueueChannel(myQueue);
    }
    
    @Bean
    HazelcastEventDrivenMessageProducer hazelcastEventDrivenMessageProducer(@Qualifier("myQueue") IQueue<Message<?>> myQueue) {
        HazelcastEventDrivenMessageProducer hazelcastEventDrivenMessageProducer = new HazelcastEventDrivenMessageProducer(myQueue);
        hazelcastEventDrivenMessageProducer.setOutputChannel(hazelcastQueueChannel(myQueue));
        hazelcastEventDrivenMessageProducer.setCacheEventTypes("ADDED");
        hazelcastEventDrivenMessageProducer.setCacheListeningPolicy(CacheListeningPolicyType.SINGLE); 
        return hazelcastEventDrivenMessageProducer;
    }

In this case i get the message and then it wrapped again and again over and over in a class called GenericMessage in my output. I abandoned this approach for now.

[artembilan]

First of all see JavaDocs of that class:

/**
 * 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 {

So, it does nothing with the data in the target Hazelacst distributedObject. It just listens to the events happened on those object.
That's why you don't see any messages for those old queue entries and you even don't see removals from the queue since this component does nothing with the data, just listens to the configured events.

That your configuration is exactly what we need to consume IQueue:

    @Bean
    IQueue<Message<?>> myQueue(@Qualifier("externalHazelcastInstance") HazelcastInstance hazelcastInstance) {
        return hazelcastInstance.getQueue(MY_TEST_QUEUE);
    }

    @Bean
    public MessageChannel hazelcastQueueChannel(@Qualifier("myQueue") IQueue<Message<?>> myQueue) {
        return new QueueChannel(myQueue);
    }

Here is a simple unit test to demonstrate an expected behavior:

@SpringJUnitConfig
@DirtiesContext
public class IQueueTests {

	@Autowired
	IQueue<Message<?>> myQueue;

	@Autowired
	PollableChannel hazelcastQueueChannel;

	@Test
	void queueChannelBackedByIQueue() {
		assertThat(this.myQueue.add(new GenericMessage<>("test1"))).isTrue();
		assertThat(this.myQueue.add(new GenericMessage<>("test2"))).isTrue();

		assertThat(this.hazelcastQueueChannel.receive(10_000))
				.isNotNull()
				.extracting(Message::getPayload)
				.isEqualTo("test1");

		assertThat(this.hazelcastQueueChannel.receive(10_000))
				.isNotNull()
				.extracting(Message::getPayload)
				.isEqualTo("test2");

		assertThat(this.myQueue.add(new GenericMessage<>("test3"))).isTrue();

		assertThat(this.hazelcastQueueChannel.receive(10_000))
				.isNotNull()
				.extracting(Message::getPayload)
				.isEqualTo("test3");

		assertThat(this.hazelcastQueueChannel.receive(10)).isNull();
		assertThat(this.myQueue).isEmpty();
	}

	@Configuration
	@EnableIntegration
	public static class ContextConfiguration {

		@Bean(destroyMethod = "shutdown")
		HazelcastInstance hazelcastInstance() {
			return Hazelcast.newHazelcastInstance();
		}

		@Bean
		IQueue<Message<?>> myQueue(HazelcastInstance hazelcastInstance) {
			return hazelcastInstance.getQueue("MY_TEST_QUEUE");
		}

		@Bean
		PollableChannel hazelcastQueueChannel(IQueue<Message<?>> myQueue) {
			return new QueueChannel(myQueue);
		}

	}

}

You just must not use HazelcastEventDrivenMessageProducer if you'd like to dequeue data from the IQueue.
That hazelcastQueueChannel can be used in a @ServiceActivator to consume the data polled from a distributed IQueue.

I'm converting this issue into a Documentation to mention such a feature we can do with Spring Integration and Hazelcast.
Plus I'll add my old sample about ITopic: https://spring.io/blog/2015/04/20/spring-integration-hazelcast-support-1-0-milestone-1-is-available

[vsalvati]

Thanks for your response. Sorry I got caught up in other things. So I am now using the setup below as I have tried before without the HazelcastEventDrivenMessageProducer class.

@Bean
    IQueue<Message<?>> myQueue(@Qualifier("externalHazelcastInstance") HazelcastInstance hazelcastInstance) {
        return hazelcastInstance.getQueue(MY_TEST_QUEUE);
    }

    @Bean
    public MessageChannel hazelcastQueueChannel(@Qualifier("myQueue") IQueue<Message<?>> myQueue) {
        return new QueueChannel(myQueue);
    }

I have my consumer using my hazelcastQueueChannel

@Slf4j
@Component
public class QueueTaskConsumer {
    @ServiceActivator(inputChannel = "hazelcastQueueChannel")
    public void readFromTaskQueue(GenericMessage<String> message) {
        log.info("received the message inside QueueTaskConsumer {} ", message);
    }
}

Here is the issue I was getting in trying this before. I schedule this remote task to add to my queue and on consumption I have a casting error.

10:10:24.400 [main] [INFO ] c.c.s.h.e.s.s.ExternalHazelcastScheduler - Scheduling queue task {"data":"Queue Test Message Data ID"} 
10:10:24.569 [scheduling-1] [ERROR] o.s.i.handler.LoggingHandler - org.springframework.messaging.MessagingException: nested exception is java.lang.ClassCastException: class java.lang.String cannot be cast to class org.springframework.messaging.Message (java.lang.String is in module java.base of loader 'bootstrap'; org.springframework.messaging.Message is in unnamed module of loader 'app')
	at org.springframework.integration.endpoint.AbstractPollingEndpoint.pollForMessage(AbstractPollingEndpoint.java:427)
	at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$createPoller$4(AbstractPollingEndpoint.java:348)
	at org.springframework.integration.util.ErrorHandlingTaskExecutor.lambda$execute$0(ErrorHandlingTaskExecutor.java:57)
	at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
	at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:55)
	at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$createPoller$5(AbstractPollingEndpoint.java:341)
	at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
	at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:95)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:829)`

When I schedule a task(a Runnable) on the remote cluster it puts it onto the queue for my consumer to read from. That code looks like this.

@Override
    @SneakyThrows
    public void run() {
        log.info("Enqueueing task with entity {} to hazelcast queue: {} ", serializedEntity, queueName);
        // IQueue<Message<?>> queue = hazelcastInstance.getQueue(queueName);  //can't do this I don't have access to Message
        IQueue<String> queue = hazelcastInstance.getQueue(queueName);
        queue.put(serializedEntity);
    }

Reminder this runs in hazelcast, a remote cluster running on my machine externally from my spring app so I don't have access to the Message class. I have to declare it as a IQueue to even get it to compile. Doing that though means I have a String on my queue and I need to cast it as a Message because of how my queue is defined and I get the following error as said before

Caused by: java.lang.ClassCastException: class java.lang.String cannot be cast to class org.springframework.messaging.Message (java.lang.String is in module java.base of loader 'bootstrap'; org.springframework.messaging.Message is in unnamed module of loader 'app')

What I need ultimately is a way to not use Spring classes in hazelcast remotely to get this to work. I've tried using user code deployment(hazelcast feature) to actually add some spring classes and that approach has failed as well.

This is why I tried using a QueueChannel like the following before.

@Bean
public MessageChannel hazelcastQueueChannel(@Qualifier("myQueue") IQueue<String> myQueue) {
  return new QueueChannel();
}

but that only consumes from the queue(no dequeue) IF i add HazelcastEventDrivenMessageProducer bean back.

Any ideas on how to procede? Can I use a SimpleMessageConverter here with Spring in some way to handle the IQueue to Message conversion?