New Source Sink Interface

ChangeMode1.java

public enum ChangelogMode {

	/**
	 * Only contains {@link ChangeType#INSERT} messages.
	 */
	INSERT_ONLY(true, false, false, false),

	/**
	 * Contains {@link ChangeType#INSERT}, {@link ChangeType#DELETE} and {@link ChangeType#UPDATE_AFTER} messages,
	 * but without {@link ChangeType#UPDATE_BEFORE}.
	 */
	UPSERET(true, true, false, true),

	/**
	 * Contains all possible {@link ChangeType} messages.
	 */
	FULL_CHANGES(true, true, true, true);

	// TODO: will introduce more modes in the future, e.g. without delete messages.

	private final boolean hasInsert;
	private final boolean hasDelete;
	private final boolean hasUpdateOld;
	private final boolean hasUpdateNew;

	private ChangelogMode(boolean hasInsert, boolean hasDelete, boolean hasUpdateOld, boolean hasUpdateNew) {
		this.hasInsert = hasInsert;
		this.hasDelete = hasDelete;
		this.hasUpdateOld = hasUpdateOld;
		this.hasUpdateNew = hasUpdateNew;
	}

	public boolean containsInsertMessages() {
		return hasInsert;
	}

	public boolean containsDeleteMessage() {
		return hasDelete;
	}

	public boolean containsUpdateBeforeMessages() {
		return hasUpdateOld;
	}

	public boolean containsUpdateAfterMessage() {
		return hasUpdateNew;
	}
}

ChangeMode2.java

/**
 * The set of changes a {@link ScanTableReader} produces and {@link TableWriter} consumes.
 */
public enum ChangeMode {

	INSERT_ONLY(
		Kind.INSERT),

	UPSERT(
		Kind.INSERT,
		Kind.UPDATE_AFTER,
		Kind.DELETE),

	ALL(
		Kind.INSERT,
		Kind.UPDATE_BEFORE,
		Kind.UPDATE_AFTER,
		Kind.DELETE);

	private final Set<Kind> kinds;

	ChangeMode(Kind firstKind, Kind... otherKinds) {
		this.kinds = Collections.unmodifiableSet(EnumSet.of(firstKind, otherKinds));
	}

	public Set<Kind> getSupportedKinds() {
		return kinds;
	}

	public boolean supportsKind(Kind kind) {
		return kinds.contains(kind);
	}
}