Flink源码解析[Source](三) - 旧版Data Sources详解&源码

public interface SourceFunction<T> extends Function, Serializable {
    
    void run(SourceContext<T> ctx) throws Exception;

    void cancel();

}

@Override
public void run(SourceContext<T> ctx) throws Exception {
    if (ownedTopicStarts.isEmpty()) {
        ctx.markAsTemporarilyIdle();
    }

    log.info("Source {} creating fetcher with offsets {}",
            taskIndex,
            StringUtils.join(ownedTopicStarts.entrySet()));

    // from this point forward:
    //   - 'snapshotState' will draw offsets from the fetcher,
    //     instead of being built from `subscribedPartitionsToStartOffsets`
    //   - 'notifyCheckpointComplete' will start to do work (i.e. commit offsets to
    //     Pulsar through the fetcher, if configured to do so)

    StreamingRuntimeContext streamingRuntime = (StreamingRuntimeContext) getRuntimeContext();

    this.pulsarFetcher = createFetcher(
            ctx,
            ownedTopicStarts,
            periodicWatermarkAssigner,
            punctuatedWatermarkAssigner,
            streamingRuntime.getProcessingTimeService(),
            streamingRuntime.getExecutionConfig().getAutoWatermarkInterval(),
            getRuntimeContext().getUserCodeClassLoader(),
            streamingRuntime);

    if (!running) {
        return;
    }

    if (discoveryIntervalMillis < 0) {
        pulsarFetcher.runFetchLoop();
    } else {
        runWithTopicsDiscovery();
    }
}

@Public
public interface RichFunction extends Function {
    void open(Configuration var1) throws Exception;

    void close() throws Exception;

    RuntimeContext getRuntimeContext();

    IterationRuntimeContext getIterationRuntimeContext();

    void setRuntimeContext(RuntimeContext var1);
}

public void open(Configuration parameters) throws Exception {
    
    
    // 1. 创建PulsarMetadataReader，PulsarMetadataReader并调用discoverTopicChanges()拿到所有的topic.
    // 创建PulsarMetadataReader管理了pulsar所有元数据的访问，比如拿到一个topic下有多少个partition等。
    this.metadataReader = createMetadataReader();
    Set<String> allTopics = metadataReader.discoverTopicChanges();
    
    // 2. 创建ownedTopicStarts成员，并赋值。赋值时依赖`restoredState(<String, MessageId>)`成员变量
    // ownedTopicStarts管理了所有topic下消费的开始offset(MessageId提供)    
    //  Map<String, MessageId> ownedTopicStarts;  
    ownedTopicStarts = new HashMap<>(); 
    if(restoredState ！= null) {
      //  从restoredState中恢复offser
      // restored在`CheckpointedFunction#initializeStates()`接口中赋值，可以参考下面具体实现分析
    } else {
      //  调用offsetForEachTopic()方法根据用户的配置(StartupMode)设置为从头还是从尾开始消费
    }
}

void snapshotState(FunctionSnapshotContext context) throws Exception;

void initializeState(FunctionInitializationContext context) throws Exception;

/**
 * This method is called as a notification once a distributed checkpoint has been completed.
 * 
 * Note that any exception during this method will not cause the checkpoint to
 * fail any more.
 * 
 * @param checkpointId The ID of the checkpoint that has been completed.
 * @throws Exception
 */
void notifyCheckpointComplete(long checkpointId) throws Exception;