olalamichelle commented on code in PR #14078:
URL: https://github.com/apache/kafka/pull/14078#discussion_r1299213641
##########
clients/src/test/java/org/apache/kafka/common/security/oauthbearer/internals/secured/RefreshingHttpsJwksTest.java:
##########
@@ -195,4 +231,68 @@ public String getBody() {
return Mockito.spy(httpsJwks);
}
+ /**
+ * A mock ScheduledExecutorService just for the test. Note that this is
not a generally reusable mock as it does not
+ * implement some interfaces like scheduleWithFixedDelay, etc. And it does
not return ScheduledFuture correctly.
+ */
+ private class MockExecutorService implements MockTime.Listener {
+ private final MockTime time;
+
+ private final TreeMap<Long, List<AbstractMap.SimpleEntry<Long,
KafkaFutureImpl<Long>>>> waiters = new TreeMap<>();
+
+ public MockExecutorService(MockTime time) {
+ this.time = time;
+ time.addListener(this);
+ }
+
+ @Override
+ public synchronized void onTimeUpdated() {
+ long timeMs = time.milliseconds();
+ while (true) {
+ Map.Entry<Long, List<AbstractMap.SimpleEntry<Long,
KafkaFutureImpl<Long>>>> entry = waiters.firstEntry();
+ if ((entry == null) || (entry.getKey() > timeMs)) {
+ break;
+ }
+ for (AbstractMap.SimpleEntry<Long, KafkaFutureImpl<Long>> pair
: entry.getValue()) {
+ pair.getValue().complete(timeMs);
+ if (pair.getKey() != null) {
+ addWaiter(entry.getKey() + pair.getKey(),
pair.getKey(), pair.getValue());
+ }
+ }
+ waiters.remove(entry.getKey());
+ }
+ }
+
+ private synchronized void addWaiter(long delayMs, Long period,
KafkaFutureImpl<Long> waiter) {
+ long timeMs = time.milliseconds();
+ if (delayMs <= 0) {
+ waiter.complete(timeMs);
+ } else {
+ long triggerTimeMs = timeMs + delayMs;
+ List<AbstractMap.SimpleEntry<Long, KafkaFutureImpl<Long>>>
futures =
+ waiters.computeIfAbsent(triggerTimeMs, k -> new
ArrayList<>());
+ futures.add(new AbstractMap.SimpleEntry<>(period, waiter));
+ }
+ }
+
+ /**
+ * Internal utility function for periodic or one time refreshes.
+ *
+ * @param period null indicates one time refresh, otherwise it is
periodic.
+ */
+ public <T> ScheduledFuture<T> schedule(final Callable<T> callable,
long delayMs, Long period) {
+
+ KafkaFutureImpl<Long> waiter = new KafkaFutureImpl<>();
+ waiter.thenApply((KafkaFuture.BaseFunction<Long, Void>) now -> {
+ try {
+ callable.call();
+ } catch (Throwable ignored) {
+ }
Review Comment:
Yes it is very similar but I have to create my own mock for 2 reasons:
1. MockScheduler.schedule method does not take a period parameter which
schedules a periodical task. I can add this function to MockScheduler but then
I will also implement the corresponding addWaiter which is pretty much just put
all the MockExecutorService code into MockScheduler. I don't think it is a good
idea since it will make MockScheduler very cumbersome. Moreover, if I do that
its schedule interface cannot take a ExecutorService parameter and it needs to
take care of the execution itself inside. See #2 for more details.
2. MockScheduler is really just a scheduler that does not take care of the
execution. It just submits tasks to the executor service. The flakiness of the
test comes from the real clock-based executor service where, because of CPU
scheduling, cannot be 100% accurate on timings. We need a callback-based one in
the test to make sure it is reliable. It is just the logic (since their
principles are the same) behind a mock scheduler and a mock executor service is
the same that we have to use MockTime and be callback-based so they are very
similar.
--
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.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
