@@ -35,7 +35,12 @@ public class K8sJobExecutorConfigs {
   /** The name of the job executor. */
   public static final String K8S_JOB_EXECUTOR_NAME = "k8s";
 
-  /** The name of the Kubernetes cluster the jobs run in, which is part of the job execution id. */
+  /**
+   * The name of the Kubernetes cluster the jobs run in, which is part of the job execution id. It
+   * must be the same on every Gravitino server, and must not be changed while jobs run: the jobs
+   * submitted under another name can't be queried nor cancelled, and stay unfinished until it is
+   * changed back.
+   */
   public static final String CLUSTER = "cluster";
 
   /** The default cluster name. */
@@ -105,7 +110,11 @@ public class K8sJobExecutorConfigs {
   /** The default status cache TTL, 5 seconds. */
   public static final long DEFAULT_STATUS_CACHE_TTL_MS = 5_000L;
 
-  /** A job whose resource still has no status after this time is considered failed. */
+  /**
+   * A job whose resource still has no status after this time is considered failed, and its resource
+   * is deleted. A job whose resource has a status without any state this version can read, for
+   * example of a newer operator, stays unfinished instead, until the operator deletes the resource.
+   */
   public static final String NO_STATUS_TIMEOUT_MS = "noStatusTimeoutInMs";
 
   /** The default no status timeout, 10 minutes. */
 
@@ -20,8 +20,10 @@
 
 import com.google.common.collect.ImmutableMap;
 import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.fabric8.kubernetes.api.model.ObjectMeta;
 import java.util.Map;
 import java.util.regex.Pattern;
+import org.apache.commons.lang3.StringUtils;
 
 /** The naming and labeling of the Kubernetes resources created for the jobs. */
 public final class K8sJobResourceUtils {
@@ -92,6 +94,27 @@ public static boolean isOfJob(HasMetadata resource, long jobId) {
     return labels != null && String.valueOf(jobId).equals(labels.get(LABEL_JOB_ID));
   }
 
+  /**
+   * Returns whether the Kubernetes resource is the one Gravitino created for the job its name was
+   * made for: it is managed by Gravitino, and its job id label is the job id its name ends with.
+   * The name alone may be used by another system, or by another Gravitino deployment with the same
+   * name prefix in the namespace, for another of its jobs.
+   *
+   * @param resource the resource
+   * @return true if the resource is a job resource of Gravitino matching its name
+   */
+  public static boolean isJobResource(HasMetadata resource) {
+    ObjectMeta metadata = resource.getMetadata();
+    Map<String, String> labels = metadata == null ? null : metadata.getLabels();
+    if (labels == null || !MANAGED_BY_GRAVITINO.equals(labels.get(LABEL_MANAGED_BY))) {
+      return false;
+    }
+    String jobId = labels.get(LABEL_JOB_ID);
+    return StringUtils.isNotBlank(jobId)
+        && metadata.getName() != null
+        && metadata.getName().endsWith("-" + jobId);
+  }
+
   /**
    * Returns whether Gravitino requested to cancel the job of the Kubernetes resource.
    *
 
@@ -0,0 +1,183 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.job.k8s;
+
+import com.google.common.collect.ImmutableMap;
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+import java.util.function.LongSupplier;
+import java.util.function.Predicate;
+import javax.annotation.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Caches the resources listed in a namespace for a short time, so that querying the status of many
+ * jobs in the same namespace costs one LIST call instead of one GET call per job.
+ *
+ * <p>A resource missing from the cache isn't necessarily gone: it may have been created after the
+ * namespace was listed, or the namespace can't be listed at all. The caller must fall back to a GET
+ * call on a cache miss.
+ *
+ * @param <T> the resource type
+ */
+final class NamespacedResourceCache<T extends HasMetadata> {
+
+  private static final class Snapshot<T> {
+
+    private final long listedAtMs;
+
+    private final Map<String, T> resources;
+
+    private Snapshot(long listedAtMs, Map<String, T> resources) {
+      this.listedAtMs = listedAtMs;
+      this.resources = resources;
+    }
+  }
+
+  private static final Logger LOG = LoggerFactory.getLogger(NamespacedResourceCache.class);
+
+  private final long ttlMs;
+
+  private final Function<String, List<T>> lister;
+
+  private final Predicate<RuntimeException> isFatal;
+
+  private final LongSupplier clock;
+
+  // The listing of each namespace, being made or made. Only one thread lists a namespace at a
+  // time, the others wait for its listing. A listing is a network call, so it is made outside of
+  // the map, which only holds the namespaces listed within the TTL or being listed.
+  private final Map<String, CompletableFuture<Snapshot<T>>> snapshots = new ConcurrentHashMap<>();
+
+  /**
+   * Creates a cache.
+   *
+   * @param ttlMs how long a namespace listing is cached, caching is disabled if not positive
+   * @param lister lists the resources of a namespace
+   * @param isFatal tells the failures of the lister that getting a resource by itself would run
+   *     into as well, for example an API server that can't be reached. They are thrown to the
+   *     caller of {@link #find(String, String)} instead of being cached as an empty listing.
+   * @param clock the current time in milliseconds
+   */
+  NamespacedResourceCache(
+      long ttlMs,
+      Function<String, List<T>> lister,
+      Predicate<RuntimeException> isFatal,
+      LongSupplier clock) {
+    this.ttlMs = ttlMs;
+    this.lister = lister;
+    this.isFatal = isFatal;
+    this.clock = clock;
+  }
+
+  /**
+   * Finds a resource in the cached listing of its namespace, and lists the namespace again if the
+   * cached listing has expired. A namespace that can't be listed, for example because listing it is
+   * forbidden, isn't listed again until the cached failure expires.
+   *
+   * @param namespace the namespace of the resource
+   * @param name the name of the resource
+   * @return the resource, or empty if it isn't in the listing or the namespace can't be listed
+   * @throws RuntimeException the failure to list the namespace, if it is a fatal one
+   */
+  Optional<T> find(String namespace, String name) {
+    if (ttlMs <= 0) {
+      return Optional.empty();
+    }
+
+    CompletableFuture<Snapshot<T>> snapshot = snapshots.get(namespace);
+    if (isExpired(snapshot)) {
+      // Drop the expired listings, also of the namespaces whose jobs are all done.
+      snapshots.values().removeIf(this::isExpired);
+      CompletableFuture<Snapshot<T>> listing = new CompletableFuture<>();
+      snapshot =
+          snapshots.compute(namespace, (ns, current) -> isExpired(current) ? listing : current);
+      if (snapshot == listing) {
+        try {
+          listing.complete(snapshotOf(namespace));
+        } catch (Throwable t) {
+          // Whatever the failure, even an error, the listing must be completed and isn't cached,
+          // otherwise the next queries of the namespace would wait for it forever. The waiting
+          // callers get the failure too.
+          snapshots.remove(namespace, listing);
+          listing.completeExceptionally(t);
+          throw t;
+        }
+      }
+    }
+
+    try {
+      return Optional.ofNullable(snapshot.join().resources.get(name));
+    } catch (CompletionException e) {
+      if (e.getCause() instanceof Error) {
+        throw (Error) e.getCause();
+      }
+      throw e.getCause() instanceof RuntimeException ? (RuntimeException) e.getCause() : e;
+    }
+  }
+
+  /**
+   * Drops the cached listing of a namespace, after a resource in it is changed. A listing of it
+   * being made isn't cached either, as it may predate the change.
+   *
+   * @param namespace the namespace
+   */
+  void invalidate(String namespace) {
+    snapshots.remove(namespace);
+  }
+
+  /** Whether a listing must be made again: there is none, it failed, or it is too old. */
+  private boolean isExpired(@Nullable CompletableFuture<Snapshot<T>> snapshot) {
+    if (snapshot == null || snapshot.isCompletedExceptionally()) {
+      return true;
+    }
+    // A listing being made is waited for.
+    Snapshot<T> made = snapshot.getNow(null);
+    return made != null && clock.getAsLong() - made.listedAtMs >= ttlMs;
+  }
+
+  private Snapshot<T> snapshotOf(String namespace) {
+    long listedAtMs = clock.getAsLong();
+    try {
+      ImmutableMap.Builder<String, T> resources = ImmutableMap.builder();
+      lister.apply(namespace).forEach(r -> resources.put(r.getMetadata().getName(), r));
+      return new Snapshot<>(listedAtMs, resources.buildKeepingLast());
+    } catch (RuntimeException e) {
+      if (isFatal.test(e)) {
+        throw e;
+      }
+      // Cache the failure as an empty listing, so that the callers get the resources one by one
+      // instead of each of them failing to list the namespace again.
+      LOG.warn(
+          "Failed to list the resources in namespace {}, getting them one by one for the next {}"
+              + " ms: {}",
+          namespace,
+          ttlMs,
+          e.toString());
+      return new Snapshot<>(listedAtMs, ImmutableMap.of());
+    }
+  }
+}
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.job.k8s;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.List;
+
+/** Reads the output of a job from the log of its pod. */
+public final class PodLogUtils {
+
+  private static final int READ_BUFFER_SIZE = 8192;
+
+  // The largest array size that is safe to allocate on every JVM.
+  private static final int MAX_BUFFER_SIZE = Integer.MAX_VALUE - 8;
+
+  private PodLogUtils() {}
+
+  /**
+   * Reads the last lines of a log stream, within its last bytes. Only the last bytes are kept in
+   * memory while reading. If the kept bytes start in the middle of a line, that partial line is
+   * dropped, unless it is the only line.
+   *
+   * <p>The whole stream is read: {@code maxBytes} bounds the memory and the lines returned, not the
+   * bytes read. The caller bounds those, for example by asking Kubernetes for the last lines of a
+   * log only.
+   *
+   * @param log the log stream, which is read to its end
+   * @param maxLines the maximum number of lines to return
+   * @param maxBytes the maximum number of bytes to read from the tail of the log
+   * @return the last lines of the log
+   * @throws IOException if the log can't be read
+   */
+  public static List<String> readLastLines(InputStream log, int maxLines, int maxBytes)
+      throws IOException {
+    Preconditions.checkArgument(maxLines > 0, "maxLines must be positive");
+    Preconditions.checkArgument(maxBytes > 0, "maxBytes must be positive");
+
+    // Keep one more byte than the window, to tell whether the window starts at a line boundary.
+    // The window is capped so that the buffer, of twice its size, fits in an array.
+    int keep = (int) Math.min(maxBytes + 1L, MAX_BUFFER_SIZE / 2);
+    int window = keep - 1;
+    int maxCapacity = Math.max(2 * keep, READ_BUFFER_SIZE);
+    // The buffer grows with the log, as most logs are much shorter than the window.
+    byte[] tail = new byte[READ_BUFFER_SIZE];
+    int length = 0;
+    byte[] chunk = new byte[READ_BUFFER_SIZE];
+    int read;
+    while ((read = log.read(chunk)) >= 0) {
+      if (length + read > tail.length && tail.length < maxCapacity) {
+        long capacity = Math.max(2L * tail.length, (long) length + read);
+        tail = Arrays.copyOf(tail, (int) Math.min(capacity, maxCapacity));
+      }
+      if (length + read > tail.length) {
+        // Drop the head, keeping the last bytes of the buffer and the chunk.
+        int kept = Math.max(keep - read, 0);
+        System.arraycopy(tail, length - kept, tail, 0, kept);
+        length = kept;
+        if (read > keep) {
+          System.arraycopy(chunk, read - keep, tail, 0, keep);
+          length = keep;
+          continue;
+        }
+      }
+      System.arraycopy(chunk, 0, tail, length, read);
+      length += read;
+    }
+
+    int start = 0;
+    boolean startsAtLineBoundary = true;
+    if (length > window) {
+      start = length - window;
+      startsAtLineBoundary = tail[start - 1] == '\n';
+    }
+    // Never start in the middle of a UTF-8 character. A '\n' is never part of one.
+    while (start < length && (tail[start] & 0xC0) == 0x80) {
+      start++;
+    }
+    String content = new String(tail, start, length - start, StandardCharsets.UTF_8);
+    if (!startsAtLineBoundary) {
+      int firstNewline = content.indexOf('\n');
+      if (firstNewline >= 0 && firstNewline < content.length() - 1) {
+        content = content.substring(firstNewline + 1);
+      }
+    }
+    if (content.isEmpty()) {
+      return ImmutableList.of();
+    }
+
+    List<String> lines = Arrays.asList(content.split("\\r?\\n", -1));
+    // A trailing line terminator doesn't start another line.
+    if (lines.get(lines.size() - 1).isEmpty()) {
+      lines = lines.subList(0, lines.size() - 1);
+    }
+    // The window only holds the line terminator of a line that starts before it.
+    if (!startsAtLineBoundary && lines.size() == 1 && lines.get(0).isEmpty()) {
+      return ImmutableList.of();
+    }
+    return ImmutableList.copyOf(lines.subList(Math.max(0, lines.size() - maxLines), lines.size()));
+  }
+}