Merge pull request #558 from yue9944882/feat/thread-safety-pager

Feat: Provide thread-safety for pager

Author: k8s-ci-robot

util/src/main/java/io/kubernetes/client/pager/Pager.java

@@ -15,25 +15,24 @@
 import com.squareup.okhttp.Call;
 import io.kubernetes.client.ApiClient;
 import io.kubernetes.client.ApiException;
-import io.kubernetes.client.models.V1ListMeta;
 import io.kubernetes.client.util.Reflect;
 import io.kubernetes.client.util.exception.ObjectMetaReflectException;
 import java.io.IOException;
 import java.lang.reflect.Type;
 import java.util.Iterator;
 import java.util.function.Function;
 
-public class Pager<ApiType, ApiListType> implements Iterable<ApiType>, Iterator<ApiType> {
-  private String continueToken;
+/*
+ * Pager encapsulates kubernetes limit/continue-based list pagination into an iterator.
+ * Note that pager is thread-safe.
+ */
+public class Pager<ApiType, ApiListType> implements Iterable<ApiType> {
   private Integer limit;
   private ApiClient client;
-  private Call call;
   private Type listType;
   private Function<PagerParams, Call> listFunc;
 
   private ApiListType listObjectCurrentPage;
-  private int offsetCurrentPage;
-  private int currentPageSize;
 
   /**
    * Pagination in kubernetes list call depends on continue and limit variable
@@ -53,88 +52,101 @@ public Pager(
   }
 
   /**
-   * returns false if kubernetes server has exhausted List.
-   *
-   * @return
-   */
-  @Override
-  public boolean hasNext() {
-    if (call == null) {
-      // the first time looping over the pager
-      return Boolean.TRUE;
-    }
-    if (continueToken == null && offsetCurrentPage >= currentPageSize) {
-      return Boolean.FALSE;
-    }
-    return Boolean.TRUE;
-  }
-
-  /**
-   * returns next chunk of List. size of list depends on limit set in constructor.
+   * Iterator iterator.
    *
-   * @return Object
+   * @return the iterator
    */
-  @Override
-  public ApiType next() {
-    try {
-      if (offsetCurrentPage >= currentPageSize) {
-        call = getNextCall(limit);
-        listObjectCurrentPage = executeRequest(call);
-
-        offsetCurrentPage = 0;
-        currentPageSize = Reflect.<ApiType>getItems(listObjectCurrentPage).size();
-      }
-      return Reflect.<ApiType>getItems(listObjectCurrentPage).get(offsetCurrentPage++);
-    } catch (ApiException e) {
-      throw new RuntimeException(e.getResponseBody());
-    } catch (ObjectMetaReflectException | IOException e) {
-      throw new RuntimeException(e);
-    }
-  }
-
   @Override
   public Iterator<ApiType> iterator() {
-    this.call = null;
-    return this;
+    return new PagerIterator();
   }
 
   /** returns next list call by setting continue variable and limit */
-  private Call getNextCall(Integer nextLimit) {
-    PagerParams params = new PagerParams((nextLimit != null) ? nextLimit : limit);
-    if (continueToken != null) {
-      params.continueToken = continueToken;
-    }
+  private Call getNextCall(Integer nextLimit, String continueToken) {
+    PagerParams params = new PagerParams((nextLimit != null) ? nextLimit : limit, continueToken);
     return listFunc.apply(params);
   }
 
   /** executes the list call and sets the continue variable for next list call */
-  private ApiListType executeRequest(Call call)
-      throws IOException, ApiException, ObjectMetaReflectException {
-    ApiListType data = client.handleResponse(call.execute(), listType);
-    V1ListMeta listMetaData = Reflect.listMetadata(data);
-    continueToken = listMetaData.getContinue();
-    return data;
+  private ApiListType executeRequest(Call call) throws IOException, ApiException {
+    return client.handleResponse(call.execute(), listType);
   }
 
   public static class PagerParams {
-    private Integer limit;
-    private String continueToken;
 
-    public PagerParams(Integer limit) {
-      this.limit = limit;
-    }
+    private String continueToken;
+    private Integer limit;
 
-    public PagerParams(Integer limit, String continueToken) {
+    private PagerParams(Integer limit, String continueToken) {
       this.limit = limit;
       this.continueToken = continueToken;
     }
 
+    /**
+     * Gets get limit.
+     *
+     * @return the get limit
+     */
     public Integer getLimit() {
       return limit;
     }
 
+    /**
+     * Gets get continue token.
+     *
+     * @return the get continue token
+     */
     public String getContinueToken() {
       return continueToken;
     }
   }
+
+  private class PagerIterator implements Iterator<ApiType> {
+
+    private boolean started;
+    private String continueToken;
+    private Call call;
+    private int offsetCurrentPage;
+    private int currentPageSize;
+
+    /**
+     * returns false if kubernetes server has exhausted List.
+     *
+     * @return the boolean
+     */
+    @Override
+    public boolean hasNext() {
+      if (!started) {
+        started = true;
+        return Boolean.TRUE;
+      }
+      return !(continueToken == null && offsetCurrentPage >= currentPageSize);
+    }
+
+    /**
+     * returns next chunk of List. size of list depends on limit set in constructor.
+     *
+     * @return the next Object
+     */
+    @Override
+    public ApiType next() {
+      try {
+        if (offsetCurrentPage >= currentPageSize) {
+
+          call = getNextCall(limit, continueToken);
+
+          listObjectCurrentPage = executeRequest(call);
+          continueToken = Reflect.listMetadata(listObjectCurrentPage).getContinue();
+
+          offsetCurrentPage = 0;
+          currentPageSize = Reflect.<ApiType>getItems(listObjectCurrentPage).size();
+        }
+        return Reflect.<ApiType>getItems(listObjectCurrentPage).get(offsetCurrentPage++);
+      } catch (ApiException e) {
+        throw new RuntimeException(e.getResponseBody());
+      } catch (ObjectMetaReflectException | IOException e) {
+        throw new RuntimeException(e);
+      }
+    }
+  }
 }

util/src/test/java/io/kubernetes/client/pager/PagerTest.java

@@ -19,7 +19,7 @@
 import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
 import static com.github.tomakehurst.wiremock.client.WireMock.urlPathEqualTo;
 import static com.github.tomakehurst.wiremock.client.WireMock.verify;
-import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.*;
 
 import com.github.tomakehurst.wiremock.junit.WireMockRule;
 import com.google.common.io.Resources;
@@ -31,7 +31,10 @@
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Paths;
-import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
@@ -56,7 +59,7 @@ public void setup() throws IOException {
   }
 
   @Test
-  public void testPaginationForNamespaceListWithSuccess() throws IOException {
+  public void testPaginationForNamespaceListWithSuccessThreadSafely() throws IOException {
     String namespaceListPage1Str = new String(Files.readAllBytes(Paths.get(LIST_PAGE1_FILE_PATH)));
     String namespaceListPage2Str = new String(Files.readAllBytes(Paths.get(LIST_PAGE2_FILE_PATH)));
     CoreV1Api api = new CoreV1Api(client);
@@ -79,6 +82,10 @@ public void testPaginationForNamespaceListWithSuccess() throws IOException {
                     .withHeader("Content-Type", "application/json")
                     .withBody(namespaceListPage2Str)));
 
+    int threads = 10;
+    CountDownLatch latch = new CountDownLatch(threads);
+    ExecutorService service = Executors.newFixedThreadPool(threads);
+
     Pager<V1Namespace, V1NamespaceList> pager =
         new Pager<V1Namespace, V1NamespaceList>(
             (Pager.PagerParams param) -> {
@@ -103,15 +110,27 @@ public void testPaginationForNamespaceListWithSuccess() throws IOException {
             1,
             V1NamespaceList.class);
 
-    int size = 0;
-    for (V1Namespace namespace : pager) {
-      assertEquals("default", namespace.getMetadata().getName());
-      size++;
+    for (int i = 0; i < threads; i++) {
+      service.submit(
+          () -> {
+            int size = 0;
+            for (V1Namespace namespace : pager) {
+              assertEquals("default", namespace.getMetadata().getName());
+              size++;
+            }
+            assertEquals(2, size);
+            latch.countDown();
+          });
+    }
+
+    try {
+      latch.await(5, TimeUnit.SECONDS);
+    } catch (InterruptedException e) {
+      fail("timed out waiting for pager finished");
     }
-    assertEquals(2, size);
 
     verify(
-        2,
+        2 * threads,
         getRequestedFor(urlPathEqualTo("/api/v1/namespaces"))
             .withQueryParam("limit", equalTo("1")));
   }
@@ -129,7 +148,7 @@ public void testPaginationForNamespaceListWithBadTokenFailure() throws IOExcepti
                     .withStatus(400)
                     .withHeader("Content-Type", "application/json")
                     .withBody(status400Str)));
-    Pager pager =
+    Pager<V1Namespace, V1NamespaceList> pager =
         new Pager<V1Namespace, V1NamespaceList>(
             (Pager.PagerParams param) -> {
               try {
@@ -152,18 +171,16 @@ public void testPaginationForNamespaceListWithBadTokenFailure() throws IOExcepti
             client,
             1,
             V1NamespaceList.class);
-    while (pager.hasNext()) {
-      try {
-        V1NamespaceList list = (V1NamespaceList) pager.next();
-        List<V1Namespace> items = list.getItems();
-        assertEquals(1, items.size());
-        for (V1Namespace namespace : items) {
-          assertEquals("default", namespace.getMetadata().getName());
-        }
-      } catch (Exception e) {
-        assertEquals(status400Str, e.getMessage());
+    int count = 0;
+    try {
+      for (V1Namespace namespace : pager) {
+        assertEquals("default", namespace.getMetadata().getName());
+        count++;
       }
+    } catch (Exception e) {
+      assertEquals(status400Str, e.getMessage());
     }
+
     verify(
         getRequestedFor(urlPathEqualTo("/api/v1/namespaces"))
             .withQueryParam("limit", equalTo("1")));
@@ -183,7 +200,7 @@ public void testPaginationForNamespaceListWithFieldSelectorFailure() throws IOEx
                     .withStatus(400)
                     .withHeader("Content-Type", "application/json")
                     .withBody(status400Str)));
-    Pager pager =
+    Pager<V1Namespace, V1NamespaceList> pager =
         new Pager<V1Namespace, V1NamespaceList>(
             (Pager.PagerParams param) -> {
               try {
@@ -206,18 +223,16 @@ public void testPaginationForNamespaceListWithFieldSelectorFailure() throws IOEx
             client,
             1,
             V1NamespaceList.class);
-    while (pager.hasNext()) {
-      try {
-        V1NamespaceList list = (V1NamespaceList) pager.next();
-        List<V1Namespace> items = list.getItems();
-        assertEquals(1, items.size());
-        for (V1Namespace namespace : items) {
-          assertEquals("default", namespace.getMetadata().getName());
-        }
-      } catch (Exception e) {
-        assertEquals(status400Str, e.getMessage());
+    int count = 0;
+    try {
+      for (V1Namespace namespace : pager) {
+        count++;
+        assertEquals("default", namespace.getMetadata().getName());
       }
+    } catch (Exception e) {
+      assertEquals(status400Str, e.getMessage());
     }
+
     verify(
         getRequestedFor(urlPathEqualTo("/api/v1/namespaces"))
             .withQueryParam("fieldSelector", equalTo("metadata.name=default"))