FlinkPulsarSource extends RichParallelSourceFunction<T> implements ResultTypeQueryable<T>,
        CheckpointListener,
        CheckpointedFunction { public FlinkPulsarSource<T> assignTimestampsAndWatermarks(AssignerWithPunctuatedWatermarks<T> assigner) { checkNotNull(assigner); if (this.periodicWatermarkAssigner != null) { throw new IllegalStateException("A periodic watermark emitter has already been set."); } try { ClosureCleaner.clean(assigner, ExecutionConfig.ClosureCleanerLevel.RECURSIVE, true); this.punctuatedWatermarkAssigner = new SerializedValue<>(assigner); return this; } catch (Exception e) { throw new IllegalArgumentException("The given assigner is not serializable", e); } } 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 @SuppressWarnings("unchecked") public void testEitherWatermarkExtractor() { try { new DummyFlinkPulsarSource<String>() .assignTimestampsAndWatermarks((AssignerWithPeriodicWatermarks<String>) null); fail(); } catch (NullPointerException ignored) { } try { new DummyFlinkPulsarSource<String>() .assignTimestampsAndWatermarks((AssignerWithPunctuatedWatermarks<String>) null); fail(); } catch (NullPointerException ignored) { } final AssignerWithPeriodicWatermarks<String> periodicAssigner = mock(AssignerWithPeriodicWatermarks.class); final AssignerWithPunctuatedWatermarks<String> punctuatedAssigner = mock(AssignerWithPunctuatedWatermarks.class); DummyFlinkPulsarSource<String> c1 = new DummyFlinkPulsarSource<>(); c1.assignTimestampsAndWatermarks(periodicAssigner); try { c1.assignTimestampsAndWatermarks(punctuatedAssigner); fail(); } catch (IllegalStateException ignored) { } DummyFlinkPulsarSource<String> c2 = new DummyFlinkPulsarSource<>(); c2.assignTimestampsAndWatermarks(punctuatedAssigner); try { c2.assignTimestampsAndWatermarks(periodicAssigner); fail(); } catch (IllegalStateException ignored) { } }