NIFI-11107 In ConsumeIMAP and ConsumePOP3 added support for OAUTH based authorization.

[tpalfy]

Summary

NIFI-11107

Tracking

Please complete the following tracking steps prior to pull request creation.

Issue Tracking

• Apache NiFi Jira issue created

Pull Request Tracking

• Pull Request title starts with Apache NiFi Jira issue number, such as NIFI-00000
• Pull Request commit message starts with Apache NiFi Jira issue number, as such NIFI-00000

Pull Request Formatting

• Pull Request based on current revision of the main branch
• Pull Request refers to a feature branch with one commit containing changes

Verification

Please indicate the verification steps performed prior to pull request creation.

Build

• Build completed using mvn clean install -P contrib-check
  • JDK 8
  • JDK 11
  • JDK 17

Licensing

• New dependencies are compatible with the Apache License 2.0 according to the License Policy
• New dependencies are documented in applicable LICENSE and NOTICE files

Documentation

• Documentation formatting appears as expected in rendered files

[nandorsoma]

Thanks @tpalfy, LGTM!
Tested ConsumeIMAP with Office365 and it works as expected.

[turcsanyip]

Thanks for your review and test @nandorsoma!
Merging to main...

[araujf]

The refresh authentication token is working? When my token is expired, a authenticate error is returned.

[AnTu2702]

I debugged the AbstractEmailProcessor.java and it seems, that AbstractEmailProcessor.java does not handle token expiration at all. The StandardOAuth2AccessTokenProvider.java does, but it would need the AbstractEmailProcessor.java to call the:

oauth2AccessTokenProvider.getAccessDetails()

in the onTrigger() method instead the onSchedule() method to do so.

The AbstractEmailProcessor.java should be fixed to handle expired OAuth2 access tokens!

[AnTu2702]

Simple add line 241 here...

[tpalfy]

@AnTu2702 The oauth2AccessTokenProviderOptional field is used in the buildUrl method where we call oauth2AccessTokenProvider.getAccessDetails().getAccessToken(). The getAccessDetail() should take care of refreshing/reacquiring tokens (this is the same call you suggest doing at the beginning of onTrigger).

So according to this, token expiration is handled already. Do we have proof that this doesn't work for some reason?

[RomanOttGmx]

Yes, implement ConsumeImapProcessor with OAUTH2 and an associated oauth2AccessTokenProviderService.
This Provider is only triggered on the onScheduled Method of the AbstractEmailProcessor and oauth2AccessTokenProvider.getAccessDetails() to refresh the token is only requested also on onScheduledMethod.
But in onTriggered Method this will not trigger the oauth2AccessTokenProvider.getAccessDetails()

We startet a local debug with break points. Configured the ConsumeImapProcessor with Schedule "Cron-Based" every 5 minutes.

onScheduled => triggered on enabling the Processor

Every 5 minutes there is the onTriggered Method triggered and not the onScheduled. That is general for all Processors which use the AbstractEmailProcessor.

[RomanOttGmx]

In IMAP Processor buildUrl will be triggered on creation of new receiver.

`protected ImapMailReceiver buildMessageReceiver(ProcessContext processContext) {
        ImapMailReceiver receiver = new ImapMailReceiver(this.buildUrl(processContext));`

And the buildMessageReceiver is requested here in AbstractEmailProcessor

 private synchronized void initializeIfNecessary(ProcessContext context, ProcessSession processSession) {
        if (this.messageReceiver == null) {
            this.processSession = processSession;
            this.messageReceiver = this.buildMessageReceiver(context);

This is called in onTrigger method of AbstractEmailProcessor, but the receiver still exists since start up, so buildUrl will never be requested again after enabling the ConsumeImapProcessor.

public void onTrigger(ProcessContext context, ProcessSession processSession) throws ProcessException {
       this.initializeIfNecessary(context, processSession);

[AnTu2702]

@tpalfy @RomanOttGmx

Setup:

Overview:

ConsumeIMAP Processor:

StandardOAuth2AccessTokenProvider Controller Service:

Debugging:

Cyclically called every 2 minutes:

Called on processor start only:

Resulting in...

[AnTu2702]

@AnTu2702 The oauth2AccessTokenProviderOptional field is used in the buildUrl method where we call oauth2AccessTokenProvider.getAccessDetails().getAccessToken(). The getAccessDetail() should take care of refreshing/reacquiring tokens (this is the same call you suggest doing at the beginning of onTrigger).

So according to this, token expiration is handled already. Do we have proof that this doesn't work for some reason?

Hey @tpalfy.

The this.messageReceiver = this.buildMessageReceiver(context) and thus buildUrl method in line 331 is only called once. Reason is line 329: if (this.messageReceiver == null)

AbstractEmailProcessor.java:

ConsumeIMAP.java:

[AnTu2702]

I was able to build and deploy a local nifi-emal-nar-1.24.0.nar file...

...working fine with the patch!

Remark:
I configured my StandardOAuth2AccessTokenProvider.java Controller Service to ALWAYS REFRESH the access token just for debugging (Access Token Expire Time = 3600 s - Refresh Window = 3590 s).

[tpalfy]

Thanks @AnTu2702 and @RomanOttGmx for the explanaion.

As for your fix @AnTu2702 I'm a bit confused though. If we only refresh the token but don't rebuild the url I don't see how that would work. Yes, we have a fresh token but are we actually using it?

[AnTu2702]

@tpalfy You are absolutely right. I just tested the refresh mechanism [OK] (and did not wait the 3600 s until the first access token really expired [NOK]) with my intended fix that morning.

So the refreshed token nonetheless needs to be considered by the processor accessing the IMAP server!
And this happens in buildurl method.

Seems, that the real fix needs to be a bit different... involving buildurl method...
What is your suggestion?

[AnTu2702]

@tpalfy @RomanOttGmx
I tried a different solution that does seem to do the job by only extending this marked piece of code:

It is a bit ugly, because the initializeIfNecessary method will this way reinitialize the messageRecevier for OAuth2 based access on every trigger() call.
But the messageRecevier doesn't offer any other way to update its url and it simply works this way...

What do you think?

[tpalfy]

@AnTu2702 Not sure about this new approach.
I'm not completely against it but I feel like rebuilding only when we get an authentication error would be more porper.

Not trivial as it would require some refactoring but I think it could be worth it.

[AnTu2702]

@tpalfy How about a quick bugfix for https://issues.apache.org/jira/browse/NIFI-12859 using this new approach for NIFI 1.26 and then refactoring it afterwards? We really really depend on having that fixed in the next couple of weeks. 😕

[tpalfy]

@AnTu2702 I think I can add a not too complicated change that only recreates the messageReciever when an authentication exception is encountered in a couple of days, hopefully today.

Since there are connections involved and given the fact that onTriggers can be called very frequently, it's better to have it like this right away.

[AnTu2702]

@AnTu2702 I think I can add a not too complicated change that only recreates the messageReciever when an authentication exception is encountered in a couple of days, hopefully today.

Since there are connections involved and given the fact that onTriggers can be called very frequently, it's better to have it like this right away.

Hm... Sorry, but I have doubts about this...

@tpalfy Our access tokens e.g. are valid for 60 minutes only.
Due to my understanding, the StandardOauth2AccessTokenProvider.java is intentionally written such that an authentication exception shall never occur at all! It will always automatically refresh any token, BEFORE it will expire (using the refresh window) by just asking it for "the access token" from outside calling getAccessDetail().
It will always be valid this way...

So an authentication exception should IMHO never be intended to occur at all...

...

But simply to have a quickfix für NIFI 1.26 still better than nothing. ;-)

[tpalfy]

Hm... Sorry, but I have doubts about this...

@tpalfy Our access tokens e.g. are valid for 60 minutes only. Due to my understanding, the StandardOauth2AccessTokenProvider.java is intentionally written such that an authentication exception shall never occur at all! It will always automatically refresh any token, BEFORE it will expire (using the refresh window) by just asking it for "the access token" from outside calling getAccessDetail(). It will always be valid this way...

So an authentication exception should IMHO never be intended to occur at all...

...

But simply to have a quickfix für NIFI 1.26 still better than nothing. ;-)

Okay, that's fair. I have a PR up with my proposed changes here: #8494

In it's first state it only recreates the mail receiver if an error is encountered. I'm going to update it by keeping the first commit, adding a revert commit and adding a new one that checks if the token has been refreshed. So both approach can be checked.

[AnTu2702]

@tpalfy Hey. How are you?

First of all: Thanks for the fix/workaround in NIFI 1.26.
Although things are basically working there, we still don't seem to be through.

The ConsumeIMAP Processor is continuously logging this ERROR...
(although the mailbox can and will be processed successfully using a refreshed token...)

We still have serious doubts to take this into production because of the log being spammed and of course the fact, that we have no clue, what causes this error.

Could you help and double check your fix/workaround?
Thank you in advance...