source: class FlinkPulsarSourceTest extends RichParallelSourceFunction<T> {
  
      private static final int maxParallelism;
      public static Properties dummyProperties;
  
      @Test
      public void checkRestoredCheckpointWhenFetcherNotReady() throws Exception {
  
target:         @SuppressWarnings("unchecked") final FlinkPulsarSource<String> source = new DummyFlinkPulsarSource<>();
  
          final TestingListState<Tuple2<String, MessageId>> restoredListState = new TestingListState<>();
          setupSource(source, true, restoredListState, true, 0, 1);
  
          // snapshot before the fetcher starts running
          source.snapshotState(new StateSnapshotContextSynchronousImpl(17, 17));
  
          // ensure that the list was cleared and refilled. while this is an implementation detail, we use it here
          // to figure out that snapshotState() actually did something.
          Assert.assertTrue(restoredListState.isClearCalled());
  
          Set<Serializable> expected = new HashSet<>();
  
          for (Serializable serializable : restoredListState.get()) {
              expected.add(serializable);
          }
  
          int counter = 0;
  
          for (Serializable serializable : restoredListState.get()) {
              assertTrue(expected.contains(serializable));
              counter++;
          }
  
          assertEquals(expected.size(), counter);
      }
  
  }