KAFKA-17415: Avoid overflow of expired timestamp #17026
Conversation
|
@frankvicky could you please fix conflicts? |
frankvicky
force-pushed
the
KAFKA-17415
branch
from
60f9567 to
d3a91df
Compare
| @@ -178,6 +178,12 @@ class DelegationTokenManagerZk(config: KafkaConfig, | |||
|
|
|||
| val issueTimeStamp = time.milliseconds | |||
| val maxLifeTime = if (maxLifeTimeMs <= 0) tokenMaxLifetime else Math.min(maxLifeTimeMs, tokenMaxLifetime) | |||
|
|
|||
| if (checkTimestampOverflow(issueTimeStamp, maxLifeTime)) { | |||
| responseCallback(CreateTokenResult(owner, tokenRequester, -1, -1, -1, "", Array[Byte](), Errors.INVALID_TIMESTAMP)) | |||
There was a problem hiding this comment.
Could we set upper bound Long.MAX_VALUE instead of throwing exception? That makes Long.MAX_VALUE valid in request. WDYT?
There was a problem hiding this comment.
Make sense, this change will make api easier to use.
| @@ -292,6 +298,8 @@ class DelegationTokenManagerZk(config: KafkaConfig, | |||
| expireResponseCallback(Errors.NONE, now) | |||
| } else if (tokenInfo.maxTimestamp < now || tokenInfo.expiryTimestamp < now) { | |||
| expireResponseCallback(Errors.DELEGATION_TOKEN_EXPIRED, -1) | |||
| } else if (now > Long.MaxValue - expireLifeTimeMs) { | |||
There was a problem hiding this comment.
we should align the behavior of handling the overflow timestamp, right?
There was a problem hiding this comment.
Yes, I missed it.
Thanks for pointing out 😸
|
@frankvicky please rebase code to trigger CI again |
| val maxLifeTimeStamp = issueTimeStamp + maxLifeTime | ||
| val expiryTimeStamp = Math.min(maxLifeTimeStamp, issueTimeStamp + defaultTokenRenewTime) | ||
|
|
||
| val isOverflowed = checkTimestampOverflow(issueTimeStamp, maxLifeTime) |
There was a problem hiding this comment.
This is too complicated. Could you please add sum method?
private static long sum(long now, long duration) {
return Long.MAX_VALUE - now <= duration ? Long.MAX_VALUE : now + duration;
} long maxTimestamp = sum(now, maxLifeTime);
long expiryTimestamp = Math.min(maxTimestamp, sum(now, tokenDefaultRenewLifetimeMs));
| @@ -294,7 +299,8 @@ class DelegationTokenManagerZk(config: KafkaConfig, | |||
| expireResponseCallback(Errors.DELEGATION_TOKEN_EXPIRED, -1) | |||
| } else { | |||
| //set expiry time stamp | |||
| val expiryTimeStamp = Math.min(tokenInfo.maxTimestamp, now + expireLifeTimeMs) | |||
| val expiryTimeStamp = if (now > Long.MaxValue - expireLifeTimeMs) Long.MaxValue | |||
There was a problem hiding this comment.
val expiryTimeStamp = Math.min(tokenInfo.maxTimestamp, sum(now, expireLifeTimeMs))
| long expiryTimestamp = Math.min(myTokenInformation.maxTimestamp(), | ||
| now + requestData.expiryTimePeriodMs()); | ||
| } else { | ||
| long expiryTimestamp = now > Long.MAX_VALUE - requestData.expiryTimePeriodMs() |
There was a problem hiding this comment.
long expiryTimestamp = Math.min(myTokenInformation.maxTimestamp(), sum(now, requestData.expiryTimePeriodMs()));
| @@ -192,6 +193,10 @@ class DelegationTokenManagerZk(config: KafkaConfig, | |||
| } | |||
| } | |||
|
|
|||
| private def sum(now: Long, duration: Long): Long = { | |||
There was a problem hiding this comment.
the zk is dropping and we don't add tests for zk, so could you please revert those changes?
There was a problem hiding this comment.
@frankvicky thanks for this patch. one small question remains.
| @@ -633,7 +633,7 @@ class SaslSslAdminIntegrationTest extends BaseAdminIntegrationTest with SaslSetu | |||
| @ValueSource(strings = Array("kraft")) | |||
| def testExpireDelegationToken(quorum: String): Unit = { | |||
| client = createAdminClient | |||
| val createDelegationTokenOptions = new CreateDelegationTokenOptions() | |||
| val createDelegationTokenOptions = new CreateDelegationTokenOptions().maxlifeTimeMs(5000) | |||
There was a problem hiding this comment.
This change was made because the previous value of maxlifeTimeMs was 5000 milliseconds. Later, I modified the config to Long.MaxValue to test overflow scenarios. To avoid breaking the original test logic, I changed the CreateDelegationTokenOptions that used the default value to explicitly set .maxlifeTimeMs(5000).
|
I have updated the description due to removing zk. |
Both ZK and KRaft modes do not handle overflow, so setting a large max lifetime results in a negative expired timestamp and negative max timestamp, which is unexpected behavior. In this PR, we are only fixing the KRaft code since ZK will be removed soon. Reviewers: Chia-Ping Tsai <[email protected]>
JIRA: KAFKA-17415
Committer Checklist (excluded from commit message)