Repository: apache/flink
File: flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TableTestStep.java (Java)

Code before:
         * Unless the test requires a very specific configuration, try to avoid calling this method
         * and fill in options later via {@link TableTestStep#apply(TableEnvironment, Map)}.
         */
        public <T> SpecificBuilder addOption(ConfigOption<T> option, String value) {
            this.options.put(option.key(), ConfigurationUtils.convertValue(value, String.class));
            return (SpecificBuilder) this;
        }

        public SpecificBuilder addDistribution(@Nullable TableDistribution distribution) {
            this.distribution = distribution;
            return (SpecificBuilder) this;
        }

        public SpecificBuilder addPartitionKeys(String... partitionKeys) {
            this.partitionKeys.addAll(Arrays.asList(partitionKeys));
            return (SpecificBuilder) this;
        }

        public SpecificBuilder enforceAppendOnly() {
            this.options.put("connector", "values");
            this.options.put("changelog-mode", "I");
            this.options.put("sink-changelog-mode-enforced", "I");
            return (SpecificBuilder) this;
        }

        public SpecificBuilder enforceUpsert() {
            return enforceUpsert(false);
        }

        public SpecificBuilder enforceUpsert(boolean withPartialDeleteOnly) {
            this.options.put("connector", "values");
            this.options.put("changelog-mode", "I,UA,D");
            this.options.put("sink-changelog-mode-enforced", "I,UA,D");
            final String withPartialDeleteOnlyStr =
                    Boolean.valueOf(withPartialDeleteOnly).toString();
            this.options.put("source.produces-delete-by-key", withPartialDeleteOnlyStr);
            this.options.put("sink.supports-delete-by-key", withPartialDeleteOnlyStr);
            return (SpecificBuilder) this;
        }

        public SpecificBuilder enforceRetract() {
            return enforceRetract(false);
        }

        public SpecificBuilder enforceRetract(boolean withPartialDeleteOnly) {
            this.options.put("connector", "values");
            this.options.put("changelog-mode", "I,UB,UA,D");
            this.options.put("sink-changelog-mode-enforced", "I,UB,UA,D");
            final String withPartialDeleteOnlyStr =
                    Boolean.valueOf(withPartialDeleteOnly).toString();
            this.options.put("source.produces-delete-by-key", withPartialDeleteOnlyStr);

Review comment (suggestion):
How about we change the methods to `inMode(ChangelogMode)`. Then one can use all the prepared methods from ChangelogMode: `builder.inMode(ChangelogMode.all())`.

Code after:
        public SpecificBuilder addOptions(Map<String, String> options) {
            this.options.putAll(options);
            return (SpecificBuilder) this;
        }

        /**
         * Unless the test requires a very specific configuration, try to avoid calling this method
         * and fill in options later via {@link TableTestStep#apply(TableEnvironment, Map)}.
         */
        public SpecificBuilder addOption(String key, String value) {
            this.options.put(key, value);
            return (SpecificBuilder) this;
        }

        /**
         * Unless the test requires a very specific configuration, try to avoid calling this method
         * and fill in options later via {@link TableTestStep#apply(TableEnvironment, Map)}.
         */
        public <T> SpecificBuilder addOption(ConfigOption<T> option, String value) {
            this.options.put(option.key(), ConfigurationUtils.convertValue(value, String.class));
            return (SpecificBuilder) this;
        }

        public SpecificBuilder addDistribution(@Nullable TableDistribution distribution) {
            this.distribution = distribution;
            return (SpecificBuilder) this;
        }

        public SpecificBuilder addPartitionKeys(String... partitionKeys) {
            this.partitionKeys.addAll(Arrays.asList(partitionKeys));
            return (SpecificBuilder) this;
        }

        public SpecificBuilder addMode(ChangelogMode mode) {
            this.options.put("connector", "values");
            final String changelogsStr =
                    mode.getContainedKinds().stream()
                            .map(ROWKIND_TO_STRING_MAP::get)
                            .collect(Collectors.joining(","));
            this.options.put("changelog-mode", changelogsStr);
            this.options.put("sink-changelog-mode-enforced", changelogsStr);
            final String keyOnlyDeleteStr = Boolean.valueOf(mode.keyOnlyDeletes()).toString();
            this.options.put("source.produces-delete-by-key", keyOnlyDeleteStr);
            this.options.put("sink.supports-delete-by-key", keyOnlyDeleteStr);
            return (SpecificBuilder) this;
        }
    }
}