flow-client: Make sure to remove any potentially expired JWT from the client used to exchange a refresh token for an access token in refresh_authorizations()
#1725
+12
−7
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
… client used to exchange a refresh token for an access token in `refresh_authorizations()`
While running down some issues with Dekaf, namely frequent consumer group rebalances, I noticed that these rebalances correlated with Dekaf errors such as:
```
dekaf: error=failed to obtain access token
Caused by:
Unauthorized: {"code":"PGRST301","details":null,"hint":null,"message":"JWT expired"}
```
jgraettinger
approved these changes
Oct 21, 2024
crates/flow-client/src/client.rs
Outdated
| let Response { | ||
| access_token, | ||
| refresh_token: next_refresh_token, | ||
| } = api_exec::<Response>(client.rpc( | ||
| } = api_exec::<Response>(client.clone().as_anonymous().rpc( |
There was a problem hiding this comment.
nit: couldn't this just be with_creds(None) ?
I also think that method should be renamed with_user_access_token() to match new() and the field itself...
There was a problem hiding this comment.
I also think that method should be renamed with_user_access_token() to match new() and the field itself...
Yup, now that the client doesn't have a refresh token, that sounds right to me
nit: couldn't this just be with_creds(None) ?
Yes. I don't remember why with_creds does user_access_token: user_access_token.or(self.user_access_token), but none of the usages of it currently pass anything other than Some(..), so I think it's safe to make it simply return
Self {
user_access_token,
..self
}…r represent its behavior
jgraettinger
approved these changes
Oct 21, 2024
jshearer
added a commit
that referenced
this pull request
Oct 21, 2024
This is a follow-up to #1725. I was still seeing some JWT expired errors, this time they looked like this: ``` WARN serve{...}: dekaf: error=status: Unauthenticated, message: "parsing jwt: token is expired by 12m4.003303197s", details: [], metadata: MetadataMap { headers: {} } ``` After some more digging, I realized that a sufficiently long-lived session containing a sufficiently long-lived `Read` could run into this issue, as the Read's journal client is never refreshed. This fixes that.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
While running down some issues with Dekaf, namely frequent consumer group rebalances, I noticed that these rebalances correlated with Dekaf errors such as:
I then realized that Dekaf was passing a client containing an expired access token to
refresh_authorizations. Sincegenerate_access_tokenRPC performs its own authentication internally by way of the provided refresh token, it shouldn't be called with an authorization header.This isn't a critical bug as it should just disconnect the session, causing a new connection to be created that'll have a freshly authenticated client, but nevertheless it's good to fix.
This change is