AbstractConfigurableMongoDbMessageStore's getNextId might overflow to long leading to "java.lang.Long cannot be cast to java.lang.Integer"

[juanavelez]

In what version(s) of Spring Integration are you seeing this issue?

It happens from version 4.3.12.RELEASE up to current. But it is possible that it happens with prior/future versions as well.

Describe the bug

Because getNextId performs an "increment" (by 1) on the "sequence" field of the special "messagesSequence" document, after Integer.MAX_VALUE calls to getNextId, next value will overflow from Integer.MAX_VALUE to a Long and when trying to retrieve it, will get fail with:

Caused by: java.lang.ClassCastException: java.lang.Long cannot be cast to java.lang.Integer

	/**
	 * Perform MongoDB {@code INC} operation for the document, which contains the {@link MessageDocument}
	 * {@code sequence}, and return the new incremented value for the new {@link MessageDocument}.
	 * The {@link #SEQUENCE_NAME} document is created on demand.
	 * @return the next sequence value.
	 */
	protected int getNextId() {
		Query query = Query.query(Criteria.where("_id").is(SEQUENCE_NAME));
		query.fields().include(MessageDocumentFields.SEQUENCE);
		return (Integer) this.mongoTemplate.findAndModify(query,
				new Update().inc(MessageDocumentFields.SEQUENCE, 1),
				FindAndModifyOptions.options().returnNew(true).upsert(true),
				Map.class, this.collectionName)
				.get(MessageDocumentFields.SEQUENCE);
	}

To Reproduce

Add Integer.MAX_VALUE + 1 of messages in aggregation store.

Caused by: java.lang.ClassCastException: java.lang.Long cannot be cast to java.lang.Integer
	at org.springframework.integration.mongodb.store.AbstractConfigurableMongoDbMessageStore.getNextId(AbstractConfigurableMongoDbMessageStore.java:180)
	at org.springframework.integration.mongodb.store.ConfigurableMongoDbMessageStore.addMessagesToGroup(ConfigurableMongoDbMessageStore.java:196)
	at org.springframework.integration.mongodb.store.ConfigurableMongoDbMessageStore.addMessageToGroup(ConfigurableMongoDbMessageStore.java:167)
	at org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler.store(AbstractCorrelatingMessageHandler.java:621)
	at org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler.handleMessageInternal(AbstractCorrelatingMessageHandler.java:413)
	at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127)

Expected behavior

Well, getNextId returns an int which populates the MessageDocument's sequence which is an int, so most likely, the MessageDocument's sequence's type needs to be changed from int to long and then getNextId, needs to return a long.

[garyrussell]

Is this a real problem, or hypothetical?

[artembilan]

Does it work if you change it to long?
Probably it is not a big deal for end-users if that property is going to be changed to long...

We might not have it as back-ported to all supported versions because existing documents in DB might not be updated to a new type, but we definitely can consider it as a fix for the current version in progress.

Thanks

[garyrussell]

But my question is:

Add Integer.MAX_VALUE + 1 of messages in aggregation store.

Is this even practical? How would such a message store perform?

[garyrussell]

Or is this just like a sequence - it doesn't mean there would be that number of messages concurrently (in which case, yes, it should be changed - but I would still expect it to take a loooong time to hit that limit).

[juanavelez]

yes, this is a real problem. we use the aggregation store to "aggregate" millions of messages sent by our customers and then release when they are complete. Most messages are single-messages but because we process 30+ (and growing) million every day we started seeing these errors starting on Friday last week For now, we are resetting that field back to 0 in MongoDB.

[juanavelez]

Or is this just like a sequence - it doesn't mean there would be that number of messages concurrently (in which case, yes, it should be changed - but I would still expect it to take a loooong time to hit that limit).

yes it is a sequence that gets incremented with each message that ts "aggreagated". For us, it took several months to reach the problem and we have a workaround (reset in MongoDB to 0) but we'd like a long term solution.

[juanavelez]

A clarification: The amount of messages that get aggregated are never beyond of the hundreds, and this issue is not related to the amount of messages in a group. The issue here is that each time a message is "aggregated", there is a mongodb-based sequence that gets incremented. You'd think that 2^31-1 would be a lot of messages to aggregate (and they are indeed a lot) but we process that much in weeks, so the use case is applicable and valid at least for us.

[artembilan]

So, yeah... Let's go ahead and fix the issue changing from int to long!
Looks like Java long is mapped to proper MongoDB type:

if (source instanceof Integer) {
	return new BsonInt32((Integer) source);
}

if (source instanceof Long) {
	return new BsonInt64((Long) source);
}