Flink FileSystemTableSink: Code Walkthrough

Source Code Analysis - How Custom Partition Commit Policies Are Created and Invoked in Flink’s FileSystemTableSink

When writing to a filesystem via Flink’s Table API or SQL with custom partitioning enabled, how does the system inject and apply the sink.partition-commit.policy.kind parameter to control partition commit behavior? This article walks through the call path and implementation details from a source-code perspective.

My own implementation, [FLINK-32388] Add the ability to pass parameters to CUSTOM PartitionCommitPolicy (https://github.com/apache/flink/pull/22831), is also based on the source code analysis in this article.

Example:

# Configure a dual-policy chain (metadata commit + custom)
sink.partition-commit.policy.kind: custom

# Custom policy configuration
sink.partition-commit.policy.custom.class: com.company.CustomPolicy
sink.partition-commit.policy.custom.parameters: param1,param2

---

1. Call Chain Overview: Where Does the Parameter Take Effect?

When a user submits a Flink SQL job, FileSystemTableSink participates in the construction of the physical execution plan as a TableSink. Its core parameter-handling logic occurs along the following call chain:

translateToPlanInternal
 └── createSinkTransformation
     └── getSinkRuntimeProvider
         └── consumeDataStream
             └── consume
                 └── createStreamingSink
                     └── sink (creates PartitionCommitter)
                         └── initializeState (builds policy chain)

---

2. Physical Plan Translation: From Plan to Sink

When the Flink planner constructs the physical execution plan (Transformation or DataStream), it invokes the translateToPlanInternal method of the TableSink. For streaming writes to a filesystem, execution eventually reaches the consume method:

@Override
public SinkRuntimeProvider getSinkRuntimeProvider(Context sinkContext) {
    return new DataStreamSinkProvider() {
        @Override
        public DataStreamSink<?> consumeDataStream(
                ProviderContext providerContext, DataStream<RowData> dataStream) {
            return consume(providerContext, dataStream, sinkContext);
        }
    };
}

Whether the job runs in streaming mode is determined by whether the data is bounded (unbounded):

• If streaming, it enters createStreamingSink to build the FileSystemSink;
• Otherwise, createBatchSink is used for batch output.

---

3. Core of the Policy Chain Construction: createStreamingSink -> sink

The most important call inside createStreamingSink is the sink() method, which is responsible for writing data to the filesystem and executing the partition commit policy to ensure data consistency and visibility.

It ultimately creates the partition commit component PartitionCommitter and connects it to the writerStream:

public static DataStreamSink<?> sink(
    ProviderContext providerContext,
    DataStream<PartitionCommitInfo> writerStream,
    Path path,
    ObjectIdentifier tableIdentifier,
    List<String> partitionKeys,
    TableMetaStoreFactory metaStoreFactory,
    FileSystemFactory fsFactory,
    ReadableConfig tableOptions
) {
    // Parameter check: whether the partition commit policy is enabled
    if (partitionKeys.size() > 0 && options.contains(SINK_PARTITION_COMMIT_POLICY_KIND)) {
        // Build the committer
        PartitionCommitter committer = new PartitionCommitter(
            path, tableIdentifier, partitionKeys, metaStoreFactory, fsFactory, tableOptions.toConfiguration()
        );
        return writerStream.transform("Partition Committer", committer);
    }
}

---

4. Runtime Policy Chain Construction: PartitionCommitter

When the Flink job starts or is restored, PartitionCommitter.initializeState() is invoked automatically:

public void initializeState(StateInitializationContext context) throws Exception {
    PartitionCommitPolicyFactory factory = new PartitionCommitPolicyFactory(
        conf.get(SINK_PARTITION_COMMIT_POLICY_KIND),     // e.g. "metastore,success-file"
        conf.get(SINK_PARTITION_COMMIT_POLICY_CLASS),     // Custom class name (optional)
        conf.get(SINK_PARTITION_COMMIT_SUCCESS_FILE_NAME), // Used by the success-file policy
        conf.get(SINK_PARTITION_COMMIT_POLICY_CLASS_PARAMETERS) // Custom class parameters (optional)
    );
    this.policies = factory.createPolicyChain(
        getUserCodeClassloader(), // User classloader
        () -> { // Factory that returns a FileSystem instance
            try {
                return fsFactory.create(locationPath.toUri());
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    );
}

---

5. Core Logic of the PartitionCommitPolicyFactory

Creating the policy chain requires the user classloader (UserCodeClassloader) for the user-defined class and a filesystem factory (FileSystem factory), ensuring that policies can correctly load user code and operate on the partition’s filesystem.

/** Create a policy chain. */
public List<PartitionCommitPolicy> createPolicyChain(
        ClassLoader cl, Supplier<FileSystem> fsSupplier) {
    String[] policyStrings = policyKind.split(",");
    return Arrays.stream(policyStrings)
            .map(
                    name -> {
                        switch (name.toLowerCase()) {
                            case PartitionCommitPolicy.METASTORE:
                                return new MetastoreCommitPolicy();
                            case PartitionCommitPolicy.SUCCESS_FILE:
                                return new SuccessFileCommitPolicy(
                                        successFileName, fsSupplier.get());
                            case PartitionCommitPolicy.CUSTOM:
                                try {
                                    if (parameters != null && !parameters.isEmpty()) {
                                        String[] paramStrings =
                                                parameters.toArray(new String[0]);
                                        Class<?>[] classes = new Class<?>[parameters.size()];
                                        for (int i = 0; i < parameters.size(); i++) {
                                            classes[i] = String.class;
                                        }
                                        return (PartitionCommitPolicy)
                                                cl.loadClass(customClass)
                                                        .getConstructor(classes)
                                                        .newInstance((Object[]) paramStrings);
                                    } else {
                                        return (PartitionCommitPolicy)
                                                cl.loadClass(customClass).newInstance();
                                    }}}}})
            .collect(Collectors.toList());}
}

💡 Key point: the custom class must implement the PartitionCommitPolicy interface and provide an appropriate constructor (either parameterized or no-arg).

---

Summary

The end-to-end injection flow for a parameter such as sink.partition-commit.policy.kind = metastore,success-file is as follows:

User configuration
  ↓
FileSystemTableSink -> sink()
  ↓
Construct PartitionCommitter
  ↓
initializeState()
  ↓
PartitionCommitPolicyFactory -> createPolicyChain()
  ↓
Create policy instances → assemble the policy chain → drive PartitionCommitter