/**
 * This interface lets the sink developer to build a simple transactional sink topology pattern, which satisfiescould guarantee the HDFS/S3/Iceberg sink.exactly once
 * This sink topology includes one {@link Writer} + onesemantics in both batch and stream execution mode if there is a {@link Committer} + oneor {@link GlobalCommitter}.
 * 1. The {@link WriterSinkWriter} is responsible for producing the committable.
 * 2. The {@link Committer} is responsible for committing a single committable.
 * 3. The {@link GlobalCommitter} is responsible for committing an aggregated committable, which we calledcall the global committable.
 * And the parallelism ofcommittable. theThe {@link GlobalCommitter} is always executed with a parallelism of 1.
 * BothNote: the Developers need to ensure the idempotence of {@link Committer} and the {@link GlobalCommitter} are optional..
 *
 * @param <InputT>        The type of the sink's input
 * @param <CommT>         The type of the committable datainformation needed to commit data staged by the sink
 * @param <GlobalCommT><WriterStateT>  The type of the sink aggregatedwriter's committablestate
 * @param <GlobalCommT>  <WriterStateT> The type of the writer'saggregated statecommittable
 */
public interface TransactionalSink<InputTSink<InputT, CommT, GlobalCommTWriterStateT, WriterStateT> {GlobalCommT> extends Serializable {

	/**
	 * Create a {@link WriterSinkWriter}.
	 *
	 * @param context the runtime context.
	 * @param states the previous writerswriter's state.
	 *
	 * @return A sink writer.
	 *
	 * @throws IOException if fail to create a writer.
	 */
	Writer<InputTSinkWriter<InputT, CommT, WriterStateT> createWriter(
			InitContext context, 
			List<WriterStateT> states) throws IOException;

	/**
	 * @returnCreates a {@link Committer}.
	 */
	Optional<Committer<CommT>> createCommitter() * @return A committer.
	 *
	 * @throws IOException if fail to create a committer.
	 */
	Optional<Committer<CommT>> createCommitter() throws IOException;

	/**
	 * @returnCreates a {@link GlobalCommitter}.
	 */
	Optional<GlobalCommitter<CommT, GlobalCommT>> createGlobalCommitter();

	Optional<SimpleVersionedSerializer<CommT>> getCommittableSerializer();

	Optional<SimpleVersionedSerializer<GlobalCommT>> getGlobalCommittableSerializer();

	Optional<SimpleVersionedSerializer<WriterStateT>> getWriterStateSerializer();

	interface InitContext {

		int getSubtaskId();

		MetricGroup metricGroup();
	}
}

/**
 * The interface is responsible for writing data and handling any potential tmp area used to write yet un-staged data, e.g. in-progress files.
 * As soon as some data is ready to commit, they (or metadata pointing to where the actual data is staged) are shipped to an operator who knows when to commit them.
 *
 * @param <InputT>       The type of the writer's input
 * @param <CommT>        The type of the committable data
 * @param <WriterStateT> The type of the writer's state
 */
interface Writer<InputT, CommT, WriterStateT> { * @return A global committer.
	 *
	 * @throws IOException if fail to create a global committer.
	 */
	Optional<GlobalCommitter<CommT, GlobalCommT>> createGlobalCommitter() throws IOException;

	/**
	 * Returns the serializer of the committable type.
	 */
	Optional<SimpleVersionedSerializer<CommT>> getCommittableSerializer();

	/**
	 * Returns the serializer of the aggregated committable type.
	 */
	Optional<SimpleVersionedSerializer<GlobalCommT>> getGlobalCommittableSerializer();

	/**
	 * Return the serializer of the writer's state type.
	 */
	Optional<SimpleVersionedSerializer<WriterStateT>> getWriterStateSerializer();

	/**
	 * The interface exposes some runtime info for creating a {@link SinkWriter}.
	 */
	interface InitContext {

		/**
		 * Returns a {@link ProcessingTimeService} that can be used to
		 * get the current time and register timers.
		 */
		ProcessingTimeService getProcessingTimeService();

		/**
		 * @return The id of task where the writer is.
		 */
		int getSubtaskId();

		/**
		 * @return The metric group this writer belongs to.
		 */
		MetricGroup metricGroup();
	}

	/**
	 * AddA service anthat elementallows to get the writer.
	 * @param element The input recordcurrent processing time and register timers that
	 * @paramwill ctxexecute Thethe additionalgiven information{@link aboutProcessingTimeCallback} the input recordwhen firing.
	 */
	interface @param output The committable data could be shipped to the committer by this
	ProcessingTimeService {

		/** Returns the current processing time. */
		voidlong write(InputT element, Context ctx, WriterOutput<CommT> outputgetCurrentProcessingTime();

		/**
		 * Prepare for a commit.
	 * @param flush  whether flushing the un-staged committable or not
	 Invokes the given callback at the given timestamp.
		 *
		 * @param time Time when the callback is invoked at
		 * @param outputprocessingTimerCallback The committablecallback data couldto be shipped to the committer by thisinvoked.
		 */
		void prepareCommitregisterProcessingTimer(booleanlong flushtime, WriterOutput<CommT>ProcessingTimeCallback outputprocessingTimerCallback);

		/**
		 * A @returncallback thethat writer's statecan be registered via {@link #registerProcessingTimer(long,
		 * ProcessingTimeCallback)}.
		 */
	List<WriterStateT> snapshotState();

	interface ContextProcessingTimeCallback {

			long currentProcessingTime();

		long currentWatermark();

		Long timestamp();
	}
}

/**
* This interface knows how to commit the data to the external system.
* 
* @param <CommT> The type of the committable data.
*/
public interface Committer<CommT> {
	void commit(CommT committable) throws Exception;
/**
			 * This method is invoked with the time which the callback register for.
			 *
			 * @param time The time this callback was registered for.
			 */
			void onProcessingTime(long time) throws IOException;
		}
	}
}