fix(tests): make event correlation timestamps deterministic

Replace Thread.sleep(100) with explicit timestamp increments in
EventCorrelatorTest. Duplicate events preserve their input timestamps,
so advancing those timestamps is sufficient to verify compression.

Aggregated events, however, generate their own timestamps using
OffsetDateTime.now(). Consecutive calls can return identical values
depending on system clock resolution, causing intermittent CI failures
on Windows.

Following the approach used by the Go client, introduce an injectable
timestamp source in EventAggregator using Supplier<OffsetDateTime>.
Add a corresponding EventCorrelator constructor to pass the supplier
through, while preserving the existing constructors and production
behaviour with OffsetDateTime::now.

Use an incrementing timestamp supplier in the test to control
aggregation timestamps independently of the system clock.

This preserves the existing timestamp assertions, removes the timing
dependency, and avoids changes to EventLogger.

Author: flrnd

extended/src/main/java/io/kubernetes/client/extended/event/legacy/EventAggregator.java

@@ -22,6 +22,8 @@
 import java.util.Objects;
 import java.util.Set;
 import java.util.function.Function;
+import java.util.function.Supplier;
+
 import org.apache.commons.lang3.tuple.MutablePair;
 
 public class EventAggregator {
@@ -30,10 +32,19 @@ public class EventAggregator {
   public static final java.time.Duration DEFAULT_EVENT_AGGREGATE_CACHE_EXPIRATION =
       java.time.Duration.ofSeconds(600);
 
+  public EventAggregator(
+     int maxLRUCacheEntries,
+     Function<CoreV1Event, MutablePair<String, String>> keyFunc,
+     Function<CoreV1Event, String> messageFunc) {
+      this(maxLRUCacheEntries, keyFunc, messageFunc, OffsetDateTime::now);
+  }
+
   public EventAggregator(
       int maxLRUCacheEntries,
       Function<CoreV1Event, MutablePair<String, String>> keyFunc,
-      Function<CoreV1Event, String> messageFunc) {
+      Function<CoreV1Event, String> messageFunc,
+      Supplier<OffsetDateTime> now) {
+    this.now = now;
     this.keyFunc = keyFunc;
     this.messageFunc = messageFunc;
     this.maxEvents = DEFAULT_MAX_EVENT_LOCAL_KEYS;
@@ -47,11 +58,12 @@ public EventAggregator(
   private Cache<String, AggregatedRecord> spammingCache;
   private Function<CoreV1Event, MutablePair<String, String>> keyFunc;
   private Function<CoreV1Event, String> messageFunc;
+  private final Supplier<OffsetDateTime> now;
 
   private final int maxEvents;
 
   public synchronized MutablePair<CoreV1Event, String> aggregate(CoreV1Event event) {
-    OffsetDateTime now = OffsetDateTime.now();
+    OffsetDateTime now = this.now.get();
 
     MutablePair<String, String> aggregatedKeys = keyFunc.apply(event);
     String aggregatedKey = aggregatedKeys.getLeft();

extended/src/main/java/io/kubernetes/client/extended/event/legacy/EventCorrelator.java

@@ -14,8 +14,12 @@
 
 import io.kubernetes.client.custom.V1Patch;
 import io.kubernetes.client.openapi.models.CoreV1Event;
+
+import java.time.OffsetDateTime;
 import java.util.Optional;
 import java.util.function.Predicate;
+import java.util.function.Supplier;
+
 import org.apache.commons.lang3.tuple.MutablePair;
 
 public class EventCorrelator {
@@ -25,14 +29,19 @@ public class EventCorrelator {
   protected EventLogger logger;
 
   public EventCorrelator() {
+    this(OffsetDateTime::now);
+  }
+
+  public EventCorrelator(Supplier<OffsetDateTime> now) {
     final int maxLRUCacheEntries = 4096;
 
     this.filter = new EventSpamFilter(maxLRUCacheEntries, EventUtils::getSpamKey)::filter;
     this.aggregator =
         new EventAggregator(
             maxLRUCacheEntries,
             EventUtils::getAggregatedAndLocalKeyByReason,
-            EventUtils::getAggregatedMessageByReason);
+            EventUtils::getAggregatedMessageByReason,
+            now);
     this.logger = new EventLogger(maxLRUCacheEntries, EventUtils::getEventKey);
   }
 

extended/src/test/java/io/kubernetes/client/extended/event/EventCorrelatorTest.java

@@ -32,6 +32,8 @@
 import java.util.Collection;
 import java.util.List;
 import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+
 import org.apache.commons.lang3.tuple.MutablePair;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.MethodSource;
@@ -143,18 +145,19 @@ void eventCorrelate(
       CoreV1Event expectedEvent,
       Boolean expectedSkip)
       throws Exception {
-    EventCorrelator correlator = new EventCorrelator();
+    AtomicReference<OffsetDateTime> time =
+    new AtomicReference<>(OffsetDateTime.now());
+    EventCorrelator correlator = new EventCorrelator(() -> time.getAndUpdate(t -> t.plusSeconds(1)));
+    OffsetDateTime now = OffsetDateTime.now();
     for (CoreV1Event event : previousEvents) {
-      OffsetDateTime now = OffsetDateTime.now();
       event.setFirstTimestamp(now);
       event.setLastTimestamp(now);
       Optional<MutablePair<CoreV1Event, V1Patch>> result = correlator.correlate(event);
       if (result.isEmpty()) {
         correlator.updateState(event);
       }
     }
-    Thread.sleep(100);
-    OffsetDateTime now = OffsetDateTime.now();
+    now = now.plusSeconds(1);
     newEvent.setFirstTimestamp(now);
     newEvent.setLastTimestamp(now);
     Optional<MutablePair<CoreV1Event, V1Patch>> result = correlator.correlate(newEvent);