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 all 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
68 changes: 31 additions & 37 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 @@ -48,8 +48,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 +85,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 +101,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 +139,7 @@ public void run() {
this.exceptionHandler.accept(apiTypeClass, t);
return;
} finally {
closeWatch();
closeWatch(watch);
}
}
} catch (ApiException e) {
Expand All @@ -185,19 +159,38 @@ public void run() {
public void stop() {
try {
isActive.set(false);
closeWatch();
} catch (Throwable t) {
this.exceptionHandler.accept(apiTypeClass, t);
}
}

private synchronized void closeWatch() throws IOException {
if (watch != null) {
watch.close();
watch = null;
private synchronized void closeWatch(Watchable<ApiType> watch) {
try {
if (watch != null) {
watch.close();
watch = null;
}
} catch (IOException e) {
log.warn("{}#Error while closing watcher", this.apiTypeClass, e);
}
}

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) {
this.store.replace(
(List<KubernetesObject>) items, resourceVersion); // down-casting is safe here
Expand Down Expand Up @@ -278,6 +271,7 @@ private void watchHandler(Watchable<ApiType> watch) {
log.debug("{}#Receiving resourceVersion {}", apiTypeClass, lastSyncResourceVersion);
}
}
closeWatch(watch);
}

static <ApiType extends KubernetesObject> void defaultWatchErrorHandler(
Expand Down