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 inMode(ChangelogMode mode) {
            this.options.put("connector", "values");
            final String changelogsStr =
                    mode.getContainedKinds().stream()
                            .map(
                                    t ->
                                            Stream.of(t.name().split("_"))
                                                    .map(name -> name.substring(0, 1))
                                                    .collect(Collectors.joining()))
                            .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;
        }
    }
}


Review comment (suggestion):
```suggestion
        public SpecificBuilder addMode(ChangelogMode mode) {
```
Sorry for the back and forth, but it seems `add` better fits to `addOption` and `addSchema`

Code after:
            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;
        }
    }
}