Skip to content

Commit b60ef97

Browse files
liunaijieliunaijie
authored
[Improve] filestore options (#8921)
1 parent 67c8b79 commit b60ef97

File tree

6 files changed

+34
-54
lines changed

6 files changed

+34
-54
lines changed

‎seatunnel-ci-tools/src/test/java/org/apache/seatunnel/api/ConnectorOptionCheckTest.java

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -205,7 +205,6 @@ private Set<String> buildWhiteList() {
205205
whiteList.add("SocketSinkOptions");
206206
whiteList.add("SelectDBSinkOptions");
207207
whiteList.add("PrometheusSinkOptions");
208-
whiteList.add("FirestoreSinkOptions");
209208
whiteList.add("MilvusSinkOptions");
210209
whiteList.add("RocketMqSourceOptions");
211210
whiteList.add("TablestoreSinkOptions");

‎seatunnel-connectors-v2/connector-google-firestore/src/main/java/org/apache/seatunnel/connectors/seatunnel/google/firestore/config/FirestoreParameters.java

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@
1717

1818
package org.apache.seatunnel.connectors.seatunnel.google.firestore.config;
1919

20-
import org.apache.seatunnel.shade.com.typesafe.config.Config;
20+
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
2121

2222
import lombok.Data;
2323

@@ -32,11 +32,11 @@ public class FirestoreParameters implements Serializable {
3232

3333
private String collection;
3434

35-
public FirestoreParameters buildWithConfig(Config config) {
36-
this.projectId = config.getString(FirestoreConfig.PROJECT_ID.key());
37-
this.collection = config.getString(FirestoreConfig.COLLECTION.key());
38-
if (config.hasPath(FirestoreConfig.CREDENTIALS.key())) {
39-
this.credentials = config.getString(FirestoreConfig.CREDENTIALS.key());
35+
public FirestoreParameters buildWithConfig(ReadonlyConfig config) {
36+
this.projectId = config.get(FirestoreSinkOptions.PROJECT_ID);
37+
this.collection = config.get(FirestoreSinkOptions.COLLECTION);
38+
if (config.getOptional(FirestoreSinkOptions.CREDENTIALS).isPresent()) {
39+
this.credentials = config.get(FirestoreSinkOptions.CREDENTIALS);
4040
}
4141
return this;
4242
}

‎seatunnel-connectors-v2/connector-google-firestore/src/main/java/org/apache/seatunnel/connectors/seatunnel/google/firestore/config/FirestoreConfig.java renamed to ‎seatunnel-connectors-v2/connector-google-firestore/src/main/java/org/apache/seatunnel/connectors/seatunnel/google/firestore/config/FirestoreSinkOptions.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@
2020
import org.apache.seatunnel.api.configuration.Option;
2121
import org.apache.seatunnel.api.configuration.Options;
2222

23-
public class FirestoreConfig {
23+
public class FirestoreSinkOptions {
2424

2525
public static final Option<String> PROJECT_ID =
2626
Options.key("project_id")

‎seatunnel-connectors-v2/connector-google-firestore/src/main/java/org/apache/seatunnel/connectors/seatunnel/google/firestore/sink/FirestoreSink.java

Lines changed: 9 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -17,70 +17,40 @@
1717

1818
package org.apache.seatunnel.connectors.seatunnel.google.firestore.sink;
1919

20-
import org.apache.seatunnel.shade.com.typesafe.config.Config;
21-
22-
import org.apache.seatunnel.api.common.PrepareFailException;
23-
import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
24-
import org.apache.seatunnel.api.sink.SeaTunnelSink;
2520
import org.apache.seatunnel.api.sink.SinkWriter;
2621
import org.apache.seatunnel.api.table.catalog.CatalogTable;
2722
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
28-
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
29-
import org.apache.seatunnel.common.config.CheckConfigUtil;
30-
import org.apache.seatunnel.common.config.CheckResult;
31-
import org.apache.seatunnel.common.constants.PluginType;
3223
import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSimpleSink;
3324
import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSinkWriter;
3425
import org.apache.seatunnel.connectors.seatunnel.google.firestore.config.FirestoreParameters;
35-
import org.apache.seatunnel.connectors.seatunnel.google.firestore.exception.FirestoreConnectorException;
36-
37-
import com.google.auto.service.AutoService;
3826

3927
import java.io.IOException;
4028
import java.util.Optional;
4129

42-
import static org.apache.seatunnel.connectors.seatunnel.google.firestore.config.FirestoreConfig.COLLECTION;
43-
import static org.apache.seatunnel.connectors.seatunnel.google.firestore.config.FirestoreConfig.PROJECT_ID;
44-
45-
@AutoService(SeaTunnelSink.class)
4630
public class FirestoreSink extends AbstractSimpleSink<SeaTunnelRow, Void> {
4731

48-
private SeaTunnelRowType rowType;
32+
private final CatalogTable catalogTable;
4933

50-
private FirestoreParameters firestoreParameters;
34+
private final FirestoreParameters firestoreParameters;
5135

52-
@Override
53-
public String getPluginName() {
54-
return "GoogleFirestore";
36+
public FirestoreSink(CatalogTable catalogTable, FirestoreParameters firestoreParameters) {
37+
this.catalogTable = catalogTable;
38+
this.firestoreParameters = firestoreParameters;
5539
}
5640

5741
@Override
58-
public void prepare(Config pluginConfig) throws PrepareFailException {
59-
CheckResult result =
60-
CheckConfigUtil.checkAllExists(pluginConfig, PROJECT_ID.key(), COLLECTION.key());
61-
if (!result.isSuccess()) {
62-
throw new FirestoreConnectorException(
63-
SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
64-
String.format(
65-
"PluginName: %s, PluginType: %s, Message: %s",
66-
getPluginName(), PluginType.SINK, result.getMsg()));
67-
}
68-
this.firestoreParameters = new FirestoreParameters().buildWithConfig(pluginConfig);
69-
}
70-
71-
@Override
72-
public void setTypeInfo(SeaTunnelRowType seaTunnelRowType) {
73-
this.rowType = seaTunnelRowType;
42+
public String getPluginName() {
43+
return "GoogleFirestore";
7444
}
7545

7646
@Override
7747
public AbstractSinkWriter<SeaTunnelRow, Void> createWriter(SinkWriter.Context context)
7848
throws IOException {
79-
return new FirestoreSinkWriter(rowType, firestoreParameters);
49+
return new FirestoreSinkWriter(catalogTable.getSeaTunnelRowType(), firestoreParameters);
8050
}
8151

8252
@Override
8353
public Optional<CatalogTable> getWriteCatalogTable() {
84-
return super.getWriteCatalogTable();
54+
return Optional.ofNullable(catalogTable);
8555
}
8656
}

‎seatunnel-connectors-v2/connector-google-firestore/src/main/java/org/apache/seatunnel/connectors/seatunnel/google/firestore/sink/FirestoreSinkFactory.java

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -18,14 +18,17 @@
1818
package org.apache.seatunnel.connectors.seatunnel.google.firestore.sink;
1919

2020
import org.apache.seatunnel.api.configuration.util.OptionRule;
21+
import org.apache.seatunnel.api.table.connector.TableSink;
2122
import org.apache.seatunnel.api.table.factory.Factory;
2223
import org.apache.seatunnel.api.table.factory.TableSinkFactory;
24+
import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext;
25+
import org.apache.seatunnel.connectors.seatunnel.google.firestore.config.FirestoreParameters;
2326

2427
import com.google.auto.service.AutoService;
2528

26-
import static org.apache.seatunnel.connectors.seatunnel.google.firestore.config.FirestoreConfig.COLLECTION;
27-
import static org.apache.seatunnel.connectors.seatunnel.google.firestore.config.FirestoreConfig.CREDENTIALS;
28-
import static org.apache.seatunnel.connectors.seatunnel.google.firestore.config.FirestoreConfig.PROJECT_ID;
29+
import static org.apache.seatunnel.connectors.seatunnel.google.firestore.config.FirestoreSinkOptions.COLLECTION;
30+
import static org.apache.seatunnel.connectors.seatunnel.google.firestore.config.FirestoreSinkOptions.CREDENTIALS;
31+
import static org.apache.seatunnel.connectors.seatunnel.google.firestore.config.FirestoreSinkOptions.PROJECT_ID;
2932

3033
@AutoService(Factory.class)
3134
public class FirestoreSinkFactory implements TableSinkFactory {
@@ -39,4 +42,12 @@ public String factoryIdentifier() {
3942
public OptionRule optionRule() {
4043
return OptionRule.builder().required(PROJECT_ID, COLLECTION).optional(CREDENTIALS).build();
4144
}
45+
46+
@Override
47+
public TableSink createSink(TableSinkFactoryContext context) {
48+
return () ->
49+
new FirestoreSink(
50+
context.getCatalogTable(),
51+
new FirestoreParameters().buildWithConfig(context.getOptions()));
52+
}
4253
}

‎seatunnel-e2e/seatunnel-connector-v2-e2e/connector-google-firestore-e2e/src/test/java/org.apache.seatunnel.e2e.connector.google.firestore/GoogleFirestoreIT.java

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@
1717

1818
package org.apache.seatunnel.e2e.connector.google.firestore;
1919

20-
import org.apache.seatunnel.connectors.seatunnel.google.firestore.config.FirestoreConfig;
20+
import org.apache.seatunnel.connectors.seatunnel.google.firestore.config.FirestoreSinkOptions;
2121
import org.apache.seatunnel.e2e.common.TestResource;
2222
import org.apache.seatunnel.e2e.common.TestSuiteBase;
2323
import org.apache.seatunnel.e2e.common.container.TestContainer;
@@ -86,9 +86,9 @@ private void initFirestoreConfig() {
8686
File file = ContainerUtil.getResourcesFile(FIRESTORE_CONF_FILE);
8787
Config config = ConfigFactory.parseFile(file);
8888
Config firestoreConfig = config.getConfig("sink").getConfig("GoogleFirestore");
89-
this.projectId = firestoreConfig.getString(FirestoreConfig.PROJECT_ID.key());
90-
this.collection = firestoreConfig.getString(FirestoreConfig.COLLECTION.key());
91-
this.credentials = firestoreConfig.getString(FirestoreConfig.CREDENTIALS.key());
89+
this.projectId = firestoreConfig.getString(FirestoreSinkOptions.PROJECT_ID.key());
90+
this.collection = firestoreConfig.getString(FirestoreSinkOptions.COLLECTION.key());
91+
this.credentials = firestoreConfig.getString(FirestoreSinkOptions.CREDENTIALS.key());
9292
}
9393

9494
@AfterAll

0 commit comments

Comments
 (0)