You are viewing an old version of this page. View the current version.

Compare with Current View Page History

« Previous Version 9 Next »


Document the state by adding a label to the FLIP page with one of "discussion", "accepted", "released", "rejected".

Discussion threadhere
Vote threadhere (<- link to https://lists.apache.org/list.html?dev@flink.apache.org)
JIRAhere (<- link to https://issues.apache.org/jira/browse/FLINK-XXXX)
ReleaseTBD

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

The goal of this FLIP is to add additional fields to Sink#InitContext to expose ExecutionConfig (or at least the ExecutionConfig#isObjectReuseEnabled) and JobID.

This is necessary to migrate the current JdbcSink to the SinkV2 (FLIP-239 -> FLINK-25421) more specifically in order to be able to reuse the current code:

  • the JdbcOutputFormat which today uses RuntimeContext to get the ExecutionConfig#isObjectReuseEnabled. 
  • the SemanticXidGenerator (for Xa-Sink) needs JobID to create the Xid.
  • the use of TypeInformation.createSerializer.


This would be necessary to another connectors as well.

Public Interfaces

  • We want to extend the Sink#InitContext to provide and  ReadableExecutionConfig and JobID.

Proposed Changes

We propose to create a new interface ReadableExecutionConfig, making ExecutionConfig extending this interface, and adding a new method to TypeInformation 

ReadableExecutionConfig
@PublicEvolving
public interface ReadableExecutionConfig extends Serializable {

    /** Returns whether object reuse has been enabled or disabled.*/
    boolean isObjectReuseEnabled();

	/** Another is/get methods needed, we should add all is/get methods from ExecutionConfig **/
    ...
}
ExecutionConfig
@Public
public class ExecutionConfig implements ReadableExecutionConfig, Archiveable<ArchivedExecutionConfig> {

	/** Nothing to be done as methods should be the same (maybe put @Override on all), its only needed to implements the new interface **/
}
TypeInformation
@Public
public abstract class TypeInformation<T> implements Serializable {
   /**
     * Creates a serializer for the type. 
	 * The serializer may use the ReadableExecutionConfig for parameterization.
     *
     * @param config The config used to parameterize the serializer.
     * @return A serializer for this type.
     */
    @PublicEvolving
    public abstract TypeSerializer<T> createSerializer(ReadableExecutionConfig config);
}

Compatibility, Deprecation, and Migration Plan

  • The change only one additional method so no impact on existing code.

Test Plan

These simple API changes will be covered by extending unit and integration tests.

Rejected Alternatives

  • Expose ExecutionConfig was reject to avoid modifications
  • Extending ExecutionConfig was reject to avoid modifications
  • No labels