Skip to content

separate initial load to avoid retaining list items in memory #4107

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 4 commits into
base: master
Choose a base branch
from
+31 −37
Open

separate initial load to avoid retaining list items in memory #4107

Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
a996bcc
separate initial load to avoid retaining list items in memory
rafaelri Jun 24, 2025
3a18c51
improve watch resource closing logic
rafaelri Jun 25, 2025
44645e2
restore import order
rafaelri Jun 25, 2025
0aa5a28
restore closeWatch
rafaelri Jun 25, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
90 changes: 43 additions & 47 deletions util/src/main/java/io/kubernetes/client/informer/cache/ReflectorRunnable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,18 @@
*/
package io.kubernetes.client.informer.cache;

import java.io.IOException;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please revert this change and maintain the orginal order of imports.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Something made VSCode change imports upon save. I ended up fixing them using vi. :(

import java.net.ConnectException;
import java.net.HttpURLConnection;
import java.time.Duration;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiConsumer;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.kubernetes.client.common.KubernetesListObject;
import io.kubernetes.client.common.KubernetesObject;
import io.kubernetes.client.informer.EventType;
Expand All @@ -23,16 +35,6 @@
import io.kubernetes.client.util.CallGeneratorParams;
import io.kubernetes.client.util.Strings;
import io.kubernetes.client.util.Watchable;
import java.io.IOException;
import java.net.ConnectException;
import java.net.HttpURLConnection;
import java.time.Duration;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ReflectorRunnable<
ApiType extends KubernetesObject, ApiListType extends KubernetesListObject>
Expand All @@ -48,8 +50,6 @@ public class ReflectorRunnable<

private boolean isLastSyncResourceVersionUnavailable;

private Watchable<ApiType> watch;

private ListerWatcher<ApiType, ApiListType> listerWatcher;

private DeltaFIFO store;
Expand Down Expand Up @@ -87,30 +87,13 @@ public void run() {
log.info("{}#Start listing and watching...", apiTypeClass);

try {
ApiListType list =
listerWatcher.list(
new CallGeneratorParams(Boolean.FALSE, getRelistResourceVersion(), null));

V1ListMeta listMeta = list.getMetadata();
String resourceVersion = listMeta.getResourceVersion();
List<? extends KubernetesObject> items = list.getItems();

if (log.isDebugEnabled()) {
log.debug("{}#Extract resourceVersion {} list meta", apiTypeClass, resourceVersion);
}
this.syncWith(items, resourceVersion);
this.lastSyncResourceVersion = resourceVersion;
this.lastSyncResourceVersion = initialLoad();
this.isLastSyncResourceVersionUnavailable = false;

Watchable<ApiType> watch = null;
if (log.isDebugEnabled()) {
log.debug("{}#Start watching with {}...", apiTypeClass, lastSyncResourceVersion);
}
while (true) {
if (!isActive.get()) {
closeWatch();
return;
}

while (isActive.get()) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I do not think that this logic is quite the same as the while(true) logic, since it will recreate the watch even if it isn't active.

try {
if (log.isDebugEnabled()) {
log.debug(
Expand All @@ -120,21 +103,14 @@ public void run() {
long jitteredWatchTimeoutSeconds =
Double.valueOf(REFLECTOR_WATCH_CLIENTSIDE_TIMEOUT.getSeconds() * (1 + Math.random()))
.longValue();
Watchable<ApiType> newWatch =
watch =
listerWatcher.watch(
new CallGeneratorParams(
Boolean.TRUE,
lastSyncResourceVersion,
Long.valueOf(jitteredWatchTimeoutSeconds).intValue()));

synchronized (this) {
if (!isActive.get()) {
newWatch.close();
continue;
}
watch = newWatch;
}
watchHandler(newWatch);
watchHandler(watch);
} catch (WatchExpiredException e) {
// Watch calls were failed due to expired resource-version. Returning
// to unwind the list-watch loops so that we can respawn a new round
Expand Down Expand Up @@ -165,7 +141,13 @@ public void run() {
this.exceptionHandler.accept(apiTypeClass, t);
return;
} finally {
closeWatch();
if (watch != null) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't see the value of this refactor, can you keep the original closeWatch code?

try {
watch.close();
} catch (IOException e) {
log.warn("{}#Error while closing watcher", this.apiTypeClass, e);
}
}
}
}
} catch (ApiException e) {
Expand All @@ -191,11 +173,20 @@ public void stop() {
}
}

private synchronized void closeWatch() throws IOException {
if (watch != null) {
watch.close();
watch = null;
}
private String initialLoad() throws ApiException {
ApiListType list =
listerWatcher.list(
new CallGeneratorParams(Boolean.FALSE, getRelistResourceVersion(), null));

V1ListMeta listMeta = list.getMetadata();
String resourceVersion = listMeta.getResourceVersion();
List<? extends KubernetesObject> items = list.getItems();

if (log.isDebugEnabled()) {
log.debug("{}#Extract resourceVersion {} list meta", apiTypeClass, resourceVersion);
}
this.syncWith(items, resourceVersion);
return resourceVersion;
}

private void syncWith(List<? extends KubernetesObject> items, String resourceVersion) {
Expand Down Expand Up @@ -278,6 +269,11 @@ private void watchHandler(Watchable<ApiType> watch) {
log.debug("{}#Receiving resourceVersion {}", apiTypeClass, lastSyncResourceVersion);
}
}
try {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why is adding this code needed?

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the changes were aimed in making it easier for the Response related resources to be reclaimed by the GC (as recommended per OkHttp that is the underlying http client used).
We've taken some heap dumps and noticed an increased number of GSON and byte[] classes held in memory and our suspicion is that the lack of a proper close and the two references held removed are the main culprit.

watch.close();
} catch (IOException e) {
log.warn("{}#Error while closing watcher", this.apiTypeClass, e);
}
}

static <ApiType extends KubernetesObject> void defaultWatchErrorHandler(
Expand Down