shashank created CAMEL-25946:
--------------------------------
Summary: camel-mongodb - after the collection of a changeStreams
consumer is dropped or renamed, the consumer retries resumeAfter the invalidate
event for as long as the route runs, and a stored invalidate token makes the
route fail to start
Key: CAMEL-25946
URL: https://issues.apache.org/jira/browse/CAMEL-25946
Project: Camel
Issue Type: Bug
Components: camel-mongodb
Reporter: shashank
Assignee: shashank
A MongoDB change stream ends with an {{invalidate}} event when its collection
is dropped or renamed, or when its database is dropped. The server then closes
the cursor (MongoDB manual, Change Events, "invalidate"). The changeStreams
consumer ({{MongoDbChangeStreamsThread}}, main 9f839f4bcc9d) routes that event
and takes its resume token like any other. When {{hasNext()}} returns false it
opens the next change stream with {{resumeAfter(token)}}. MongoDB rejects that:
{{resumeAfter}} cannot resume from an invalidate event (MongoDB manual, "Resume
a Change Stream"; server error 260 InvalidResumeToken, "Attempting to resume a
change stream using 'resumeAfter' is not allowed from an invalidate
notification."). So:
* after a drop or rename of the collection, the consumer fails to open a cursor
every {{cursorRegenerationDelay}} (1 second by default), each time logging a
WARN with a stack trace, for as long as the route runs. A collection created
again under the same name (a reload job, a test fixture) is never consumed. The
route still shows as Started;
* with {{changeStreamTokenRepository}} (or a resume strategy) the invalidate
token is stored. When the route starts again it opens the change stream with
that token, and the start fails with {{FailedToStartRouteException}} caused by
the same error, until someone deletes the stored token.
*Fix*: when opening a change stream with {{resumeAfter(token)}} fails with
InvalidResumeToken, the consumer opens it with {{startAfter(token)}} instead
(MongoDB 4.2 and later). That starts a new change stream after the invalidate
event, which sees the collection created again. The consumer logs this at INFO,
once for each change stream it opens with the invalidate token. A token that is
invalid for another reason fails with {{startAfter}} too, and the consumer
tries again after the delay, as before. The resume token of a normal event is
still used with {{resumeAfter}}.
*Tests*: new {{MongoDbChangeStreamsInvalidateTest}}. Its fake change stream
answers like a 4.2+ server on which the collection was dropped and created
again: the insert, drop and invalidate events, then a closed cursor;
{{resumeAfter}} the invalidate token throws {{MongoCommandException}} 260;
{{startAfter}} it returns the insert into the new collection. Both tests fail
on main:
* {{consumerGoesOnAfterTheCollectionIsDropped}}: {{mock://result Received
message count. Expected: <4> but was: <3>}};
* {{consumerStartsFromAStoredTokenOfAnInvalidateEvent}}:
{{FailedToStartRouteException: Failed to start route: mongo because: Command
execution failed on MongoDB server with error 260 (InvalidResumeToken)}}.
camel-mongodb unit tests with the fix: 24 run, 0 failures.
*How it was found*: library-contract mining. I listed what the driver and the
server do at each step of a change stream (an invalidate closes the cursor;
resumeAfter is rejected for that token; startAfter is not) and checked the
consumer's cursor regeneration against that.
*Affected*: main, camel-4.23.x, camel-4.22.x, camel-4.18.x and camel-4.14.x.
They all open the next change stream with {{resumeAfter(resumeToken)}}. I read
the code on those branches but did not run the tests there.
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
