|
19 | 19 |
|
20 | 20 | import org.apache.seatunnel.api.configuration.Option; |
21 | 21 | import org.apache.seatunnel.api.configuration.Options; |
22 | | -import org.apache.seatunnel.api.configuration.util.OptionRule; |
23 | 22 | import org.apache.seatunnel.api.sink.DataSaveMode; |
24 | 23 | import org.apache.seatunnel.api.sink.SaveModePlaceHolder; |
25 | 24 | import org.apache.seatunnel.api.sink.SchemaSaveMode; |
26 | 25 |
|
27 | 26 | import java.util.Map; |
28 | 27 |
|
29 | | -import static org.apache.seatunnel.api.options.SinkConnectorCommonOptions.MULTI_TABLE_SINK_REPLICA; |
30 | | -import static org.apache.seatunnel.connectors.doris.config.DorisOptions.DATABASE; |
31 | | -import static org.apache.seatunnel.connectors.doris.config.DorisOptions.DORIS_BATCH_SIZE; |
32 | | -import static org.apache.seatunnel.connectors.doris.config.DorisOptions.FENODES; |
33 | | -import static org.apache.seatunnel.connectors.doris.config.DorisOptions.PASSWORD; |
34 | | -import static org.apache.seatunnel.connectors.doris.config.DorisOptions.QUERY_PORT; |
35 | | -import static org.apache.seatunnel.connectors.doris.config.DorisOptions.TABLE; |
36 | | -import static org.apache.seatunnel.connectors.doris.config.DorisOptions.TABLE_IDENTIFIER; |
37 | | -import static org.apache.seatunnel.connectors.doris.config.DorisOptions.USERNAME; |
| 28 | +public class DorisSinkOptions extends DorisBaseOptions { |
38 | 29 |
|
39 | | -public interface DorisSinkOptions { |
40 | | - |
41 | | - int DEFAULT_SINK_CHECK_INTERVAL = 10000; |
42 | | - int DEFAULT_SINK_MAX_RETRIES = 3; |
43 | | - int DEFAULT_SINK_BUFFER_SIZE = 256 * 1024; |
44 | | - int DEFAULT_SINK_BUFFER_COUNT = 3; |
| 30 | + @Deprecated |
| 31 | + public static final Option<String> TABLE_IDENTIFIER = |
| 32 | + Options.key("table.identifier") |
| 33 | + .stringType() |
| 34 | + .noDefaultValue() |
| 35 | + .withDescription("the doris table name."); |
45 | 36 |
|
46 | | - Option<Boolean> SINK_ENABLE_2PC = |
| 37 | + public static final Option<Boolean> SINK_ENABLE_2PC = |
47 | 38 | Options.key("sink.enable-2pc") |
48 | 39 | .booleanType() |
49 | 40 | .defaultValue(false) |
50 | 41 | .withDescription("enable 2PC while loading"); |
51 | 42 |
|
52 | | - Option<Integer> SINK_CHECK_INTERVAL = |
| 43 | + public static final Option<Integer> SINK_CHECK_INTERVAL = |
53 | 44 | Options.key("sink.check-interval") |
54 | 45 | .intType() |
55 | | - .defaultValue(DEFAULT_SINK_CHECK_INTERVAL) |
| 46 | + .defaultValue(10000) |
56 | 47 | .withDescription("check exception with the interval while loading"); |
57 | | - Option<Integer> SINK_MAX_RETRIES = |
| 48 | + public static final Option<Integer> SINK_MAX_RETRIES = |
58 | 49 | Options.key("sink.max-retries") |
59 | 50 | .intType() |
60 | | - .defaultValue(DEFAULT_SINK_MAX_RETRIES) |
| 51 | + .defaultValue(3) |
61 | 52 | .withDescription("the max retry times if writing records to database failed."); |
62 | | - Option<Integer> SINK_BUFFER_SIZE = |
| 53 | + public static final Option<Integer> SINK_BUFFER_SIZE = |
63 | 54 | Options.key("sink.buffer-size") |
64 | 55 | .intType() |
65 | | - .defaultValue(DEFAULT_SINK_BUFFER_SIZE) |
| 56 | + .defaultValue(256 * 1024) |
66 | 57 | .withDescription("the buffer size to cache data for stream load."); |
67 | | - Option<Integer> SINK_BUFFER_COUNT = |
| 58 | + public static final Option<Integer> SINK_BUFFER_COUNT = |
68 | 59 | Options.key("sink.buffer-count") |
69 | 60 | .intType() |
70 | | - .defaultValue(DEFAULT_SINK_BUFFER_COUNT) |
| 61 | + .defaultValue(3) |
71 | 62 | .withDescription("the buffer count to cache data for stream load."); |
72 | | - Option<String> SINK_LABEL_PREFIX = |
| 63 | + public static final Option<String> SINK_LABEL_PREFIX = |
73 | 64 | Options.key("sink.label-prefix") |
74 | 65 | .stringType() |
75 | 66 | .defaultValue("") |
76 | 67 | .withDescription("the unique label prefix."); |
77 | | - Option<Boolean> SINK_ENABLE_DELETE = |
| 68 | + public static final Option<Boolean> SINK_ENABLE_DELETE = |
78 | 69 | Options.key("sink.enable-delete") |
79 | 70 | .booleanType() |
80 | 71 | .defaultValue(false) |
81 | 72 | .withDescription("whether to enable the delete function"); |
82 | 73 |
|
83 | | - Option<Map<String, String>> DORIS_SINK_CONFIG_PREFIX = |
| 74 | + public static final Option<Map<String, String>> DORIS_SINK_CONFIG_PREFIX = |
84 | 75 | Options.key("doris.config") |
85 | 76 | .mapType() |
86 | 77 | .noDefaultValue() |
87 | 78 | .withDescription( |
88 | 79 | "The parameter of the Stream Load data_desc. " |
89 | 80 | + "The way to specify the parameter is to add the prefix `doris.config` to the original load parameter name "); |
90 | 81 |
|
91 | | - Option<String> DEFAULT_DATABASE = |
| 82 | + public static final Option<String> DEFAULT_DATABASE = |
92 | 83 | Options.key("default-database") |
93 | 84 | .stringType() |
94 | 85 | .defaultValue("information_schema") |
95 | 86 | .withDescription(""); |
96 | 87 |
|
97 | | - Option<SchemaSaveMode> SCHEMA_SAVE_MODE = |
| 88 | + public static final Option<SchemaSaveMode> SCHEMA_SAVE_MODE = |
98 | 89 | Options.key("schema_save_mode") |
99 | 90 | .enumType(SchemaSaveMode.class) |
100 | 91 | .defaultValue(SchemaSaveMode.CREATE_SCHEMA_WHEN_NOT_EXIST) |
101 | 92 | .withDescription("schema_save_mode"); |
102 | 93 |
|
103 | | - Option<DataSaveMode> DATA_SAVE_MODE = |
| 94 | + public static final Option<DataSaveMode> DATA_SAVE_MODE = |
104 | 95 | Options.key("data_save_mode") |
105 | 96 | .enumType(DataSaveMode.class) |
106 | 97 | .defaultValue(DataSaveMode.APPEND_DATA) |
107 | 98 | .withDescription("data_save_mode"); |
108 | 99 |
|
109 | | - Option<String> CUSTOM_SQL = |
| 100 | + public static final Option<String> CUSTOM_SQL = |
110 | 101 | Options.key("custom_sql").stringType().noDefaultValue().withDescription("custom_sql"); |
111 | 102 |
|
112 | | - Option<Boolean> NEEDS_UNSUPPORTED_TYPE_CASTING = |
| 103 | + public static final Option<Boolean> NEEDS_UNSUPPORTED_TYPE_CASTING = |
113 | 104 | Options.key("needs_unsupported_type_casting") |
114 | 105 | .booleanType() |
115 | 106 | .defaultValue(false) |
116 | 107 | .withDescription( |
117 | 108 | "Whether to enable the unsupported type casting, such as Decimal64 to Double"); |
118 | 109 |
|
119 | 110 | // create table |
120 | | - Option<String> SAVE_MODE_CREATE_TEMPLATE = |
| 111 | + public static final Option<String> SAVE_MODE_CREATE_TEMPLATE = |
121 | 112 | Options.key("save_mode_create_template") |
122 | 113 | .stringType() |
123 | 114 | .defaultValue( |
@@ -147,27 +138,4 @@ public interface DorisSinkOptions { |
147 | 138 | + "\"disable_auto_compaction\" = \"false\"\n" |
148 | 139 | + ")") |
149 | 140 | .withDescription("Create table statement template, used to create Doris table"); |
150 | | - |
151 | | - OptionRule.Builder SINK_RULE = |
152 | | - OptionRule.builder() |
153 | | - .required( |
154 | | - FENODES, |
155 | | - USERNAME, |
156 | | - PASSWORD, |
157 | | - SINK_LABEL_PREFIX, |
158 | | - DORIS_SINK_CONFIG_PREFIX, |
159 | | - DATA_SAVE_MODE, |
160 | | - SCHEMA_SAVE_MODE) |
161 | | - .optional( |
162 | | - DATABASE, |
163 | | - TABLE, |
164 | | - TABLE_IDENTIFIER, |
165 | | - QUERY_PORT, |
166 | | - DORIS_BATCH_SIZE, |
167 | | - SINK_ENABLE_2PC, |
168 | | - SINK_ENABLE_DELETE, |
169 | | - MULTI_TABLE_SINK_REPLICA, |
170 | | - SAVE_MODE_CREATE_TEMPLATE, |
171 | | - NEEDS_UNSUPPORTED_TYPE_CASTING) |
172 | | - .conditional(DATA_SAVE_MODE, DataSaveMode.CUSTOM_PROCESSING, CUSTOM_SQL); |
173 | 141 | } |
0 commit comments