FlinkPulsarSource extends RichParallelSourceFunction<T> implements ResultTypeQueryable<T>,
        CheckpointListener,
        CheckpointedFunction { @Override public void snapshotState(FunctionSnapshotContext context) throws Exception { if (!running) { log.debug("snapshotState() called on closed source"); } else { unionOffsetStates.clear(); PulsarFetcher fetcher = this.pulsarFetcher; if (fetcher == null) { for (Map.Entry<TopicRange, MessageId> entry : ownedTopicStarts.entrySet()) { unionOffsetStates.add(Tuple3.of(entry.getKey(), entry.getValue(), getSubscriptionName())); } pendingOffsetsToCommit.put(context.getCheckpointId(), restoredState); } else { Map<TopicRange, MessageId> currentOffsets = fetcher.snapshotCurrentState(); pendingOffsetsToCommit.put(context.getCheckpointId(), currentOffsets); for (Map.Entry<TopicRange, MessageId> entry : currentOffsets.entrySet()) { unionOffsetStates.add(Tuple3.of(entry.getKey(), entry.getValue(), getSubscriptionName())); } while (pendingOffsetsToCommit.size() > MAX_NUM_PENDING_CHECKPOINTS) { pendingOffsetsToCommit.remove(0); } } } } FlinkPulsarSource(
            String adminUrl,
            ClientConfigurationData clientConf,
            PulsarDeserializationSchema<T> deserializer,
            Properties properties);  FlinkPulsarSource(
            String serviceUrl,
            String adminUrl,
            PulsarDeserializationSchema<T> deserializer,
            Properties properties); FlinkPulsarSource<T> assignTimestampsAndWatermarks(AssignerWithPunctuatedWatermarks<T> assigner); FlinkPulsarSource<T> assignTimestampsAndWatermarks(AssignerWithPeriodicWatermarks<T> assigner); FlinkPulsarSource<T> setStartFromEarliest(); FlinkPulsarSource<T> setStartFromLatest(); FlinkPulsarSource<T> setStartFromSpecificOffsets(Map<String, MessageId> specificStartupOffsets); FlinkPulsarSource<T> setStartFromSubscription(String externalSubscriptionName); FlinkPulsarSource<T> setStartFromSubscription(String externalSubscriptionName, MessageId subscriptionPosition); @Override void open(Configuration parameters); @Override void run(SourceContext<T> ctx); void joinDiscoveryLoopThread(); void runWithTopicsDiscovery(); @Override void close(); @Override void cancel(); @Override TypeInformation<T> getProducedType(); @Override void initializeState(FunctionInitializationContext context); @Override void snapshotState(FunctionSnapshotContext context); @Override void notifyCheckpointComplete(long checkpointId); void notifyCheckpointAborted(long checkpointId); Map<TopicRange, MessageId> offsetForEachTopic(
            Set<TopicRange> topics,
            StartupMode mode,
            Map<TopicRange, MessageId> specificStartupOffsets); LinkedMap getPendingOffsetsToCommit(); Map<TopicRange, MessageId> getOwnedTopicStarts(); static final int MAX_NUM_PENDING_CHECKPOINTS; }
@Test public void checkRestoredCheckpointWhenFetcherNotReady() throws Exception { @SuppressWarnings("unchecked") final FlinkPulsarSource<String> source = new DummyFlinkPulsarSource<>(); final TestingListState<Tuple2<String, MessageId>> restoredListState = new TestingListState<>(); setupSource(source, true, restoredListState, true, 0, 1); source.snapshotState(new StateSnapshotContextSynchronousImpl(17, 17)); 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); }