C0urante commented on a change in pull request #10014:
URL: https://github.com/apache/kafka/pull/10014#discussion_r568653512
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java
##########
@@ -381,7 +381,7 @@ public void tick() {
log.debug("Scheduled rebalance at: {} (now: {}
nextRequestTimeoutMs: {}) ",
scheduledRebalance, now, nextRequestTimeoutMs);
}
- if (internalRequestValidationEnabled() && keyExpiration <
Long.MAX_VALUE) {
+ if (isLeader() && internalRequestValidationEnabled() && keyExpiration
< Long.MAX_VALUE) {
Review comment:
There's a rationale for leaving the check in here but I'm not sold
either way. Let me know what you think:
The `Scheduled next key rotation at...` log message should ideally only be
emitted if key rotation is actually enabled. Key rotation can be [disabled by
setting the ttl to
0](https://github.com/apache/kafka/blob/821d867c1bd01b88aa695827cb57c9aa764857a4/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java#L1579).
If someone has key rotation disabled, emitting this log message doesn't make a
lot of sense (even if the scheduled rotation time is millions of years in the
future).
This also mirrors the structure of the scheduled rebalance delay check
above, although that check is more necessary because of the tweaking of the
`rebalanceResolved` field.
##########
File path:
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java
##########
@@ -2161,6 +2163,83 @@ public Boolean answer() throws Throwable {
PowerMock.verifyAll();
}
+ @Test
+ public void testKeyRotationWhenWorkerBecomesLeader() throws Exception {
+ EasyMock.expect(member.memberId()).andStubReturn("member");
+
EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V2);
+ EasyMock.expect(worker.getPlugins()).andReturn(plugins);
+
+ expectRebalance(1, Collections.emptyList(), Collections.emptyList());
+ expectPostRebalanceCatchup(SNAPSHOT);
+ // First rebalance: poll indefinitely as no key has been read yet, so
expiration doesn't come into play
+ member.poll(Long.MAX_VALUE);
+ EasyMock.expectLastCall();
+
+ expectRebalance(2, Collections.emptyList(), Collections.emptyList());
+ SessionKey initialKey = new SessionKey(EasyMock.mock(SecretKey.class),
0);
+ ClusterConfigState snapshotWithKey = new ClusterConfigState(2,
initialKey, Collections.singletonMap(CONN1, 3),
+ Collections.singletonMap(CONN1, CONN1_CONFIG),
Collections.singletonMap(CONN1, TargetState.STARTED),
+ TASK_CONFIGS_MAP, Collections.<String>emptySet());
+ expectPostRebalanceCatchup(snapshotWithKey);
+ // Second rebalance: poll indefinitely as worker is follower, so
expiration still doesn't come into play
+ member.poll(Long.MAX_VALUE);
+ EasyMock.expectLastCall();
+
+ expectRebalance(2, Collections.emptyList(), Collections.emptyList(),
"member", MEMBER_URL);
+ Capture<SessionKey> updatedKey = EasyMock.newCapture();
+ configBackingStore.putSessionKey(EasyMock.capture(updatedKey));
+ EasyMock.expectLastCall().andAnswer(() -> {
+ configUpdateListener.onSessionKeyUpdate(updatedKey.getValue());
+ return null;
+ });
+ // Third rebalance: poll for a limited time as worker has become
leader and must wake up for key expiration
+ Capture<Long> pollTimeout = EasyMock.newCapture();
+ member.poll(EasyMock.captureLong(pollTimeout));
+ EasyMock.expectLastCall();
+
+ PowerMock.replayAll();
+
+ herder.tick();
+ configUpdateListener.onSessionKeyUpdate(initialKey);
+ herder.tick();
+ herder.tick();
+
+ assertTrue(pollTimeout.getValue() <=
DistributedConfig.INTER_WORKER_KEY_TTL_MS_MS_DEFAULT);
+ }
Review comment:
There are a bunch of expectations set up with the `expectRebalance` and
`expectPostRebalance` methods; I've added the `PowerMock::verifyAll` call as a
result. Good catch, thanks!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
