This is an automated email from the ASF dual-hosted git repository.
chenguangsheng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new df0798616 [ISSUE #4454] Support File Sink Connector (#4455)
df0798616 is described below
commit df07986162cc1bb12c2d54769807af1189baeb52
Author: mike_xwm <[email protected]>
AuthorDate: Fri Sep 22 18:23:23 2023 +0800
[ISSUE #4454] Support File Sink Connector (#4455)
* [ISSUE #4454] Support File Sink Connector
* Update FileSinkConnector.java
* update openWithNewFile method
---
.../eventmesh-connector-file/build.gradle | 22 +++
.../eventmesh-connector-file/gradle.properties | 18 +++
.../connector/file/config/FileServerConfig.java | 33 ++++
.../connector/file/server/FileConnectServer.java | 47 ++++++
.../connector/file/sink/config/FileSinkConfig.java | 35 +++++
.../file/sink/config/SinkConnectorConfig.java | 28 ++++
.../file/sink/connector/FileSinkConnector.java | 167 +++++++++++++++++++++
.../file/source/config/FileSourceConfig.java | 30 ++++
.../file/source/config/SourceConnectorConfig.java | 33 ++++
.../file/source/connector/FileSourceConnector.java | 87 +++++++++++
.../src/main/resources/server-config.yml | 19 +++
.../src/main/resources/sink-config.yml | 29 ++++
.../src/main/resources/source-config.yml | 40 +++++
settings.gradle | 1 +
14 files changed, 589 insertions(+)
diff --git a/eventmesh-connectors/eventmesh-connector-file/build.gradle
b/eventmesh-connectors/eventmesh-connector-file/build.gradle
new file mode 100644
index 000000000..7688ef5e2
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-file/build.gradle
@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+dependencies {
+ api project(":eventmesh-openconnect:eventmesh-openconnect-java")
+ compileOnly 'org.projectlombok:lombok'
+ annotationProcessor 'org.projectlombok:lombok'
+}
\ No newline at end of file
diff --git a/eventmesh-connectors/eventmesh-connector-file/gradle.properties
b/eventmesh-connectors/eventmesh-connector-file/gradle.properties
new file mode 100644
index 000000000..1820ec46f
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-file/gradle.properties
@@ -0,0 +1,18 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+pluginType=connector
+pluginName=file
\ No newline at end of file
diff --git
a/eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/config/FileServerConfig.java
b/eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/config/FileServerConfig.java
new file mode 100644
index 000000000..b6b506521
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/config/FileServerConfig.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.file.config;
+
+import org.apache.eventmesh.openconnect.api.config.Config;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class FileServerConfig extends Config {
+
+ private boolean sourceEnable;
+
+ private boolean sinkEnable;
+
+}
diff --git
a/eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/server/FileConnectServer.java
b/eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/server/FileConnectServer.java
new file mode 100644
index 000000000..f134b2df6
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/server/FileConnectServer.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.file.server;
+
+import org.apache.eventmesh.connector.file.config.FileServerConfig;
+import org.apache.eventmesh.connector.file.sink.connector.FileSinkConnector;
+import
org.apache.eventmesh.connector.file.source.connector.FileSourceConnector;
+import org.apache.eventmesh.openconnect.Application;
+import org.apache.eventmesh.openconnect.util.ConfigUtil;
+
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class FileConnectServer {
+
+
+ public static void main(String[] args) throws Exception {
+
+ FileServerConfig serverConfig =
ConfigUtil.parse(FileServerConfig.class, "server-config.yml");
+
+ if (serverConfig.isSourceEnable()) {
+ Application fileSourceApp = new Application();
+ fileSourceApp.run(FileSourceConnector.class);
+ }
+
+ if (serverConfig.isSinkEnable()) {
+ Application fileSinkApp = new Application();
+ fileSinkApp.run(FileSinkConnector.class);
+ }
+ }
+}
diff --git
a/eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/sink/config/FileSinkConfig.java
b/eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/sink/config/FileSinkConfig.java
new file mode 100644
index 000000000..abdae5277
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/sink/config/FileSinkConfig.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.file.sink.config;
+
+import org.apache.eventmesh.openconnect.api.config.SinkConfig;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class FileSinkConfig extends SinkConfig {
+
+ public SinkConnectorConfig connectorConfig;
+
+ private Integer flushSize = 1000;
+
+ private boolean hourlyFlushEnabled = false;
+
+}
diff --git
a/eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/sink/config/SinkConnectorConfig.java
b/eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/sink/config/SinkConnectorConfig.java
new file mode 100644
index 000000000..a8cf958f7
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/sink/config/SinkConnectorConfig.java
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.file.sink.config;
+
+import lombok.Data;
+
+@Data
+public class SinkConnectorConfig {
+
+ private String connectorName;
+
+ private String topic;
+}
diff --git
a/eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/sink/connector/FileSinkConnector.java
b/eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/sink/connector/FileSinkConnector.java
new file mode 100644
index 000000000..89f25dd38
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/sink/connector/FileSinkConnector.java
@@ -0,0 +1,167 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.file.sink.connector;
+
+import org.apache.eventmesh.connector.file.sink.config.FileSinkConfig;
+import org.apache.eventmesh.openconnect.api.config.Config;
+import org.apache.eventmesh.openconnect.api.connector.ConnectorContext;
+import org.apache.eventmesh.openconnect.api.connector.SinkConnectorContext;
+import org.apache.eventmesh.openconnect.api.sink.Sink;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
+import java.time.LocalDateTime;
+import java.util.Calendar;
+import java.util.List;
+import java.util.Locale;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class FileSinkConnector implements Sink {
+
+ private static final AtomicInteger fileSize = new AtomicInteger(0);
+
+ private String filePath;
+
+ private String fileName;
+
+ private int flushSize;
+
+ private boolean hourlyFlushEnabled;
+
+ private FileSinkConfig sinkConfig;
+
+ private PrintStream outputStream;
+
+ @Override
+ public Class<? extends Config> configClass() {
+ return FileSinkConfig.class;
+ }
+
+ @Override
+ public void init(Config config) {
+ // init config for hdfs source connector
+ this.sinkConfig = (FileSinkConfig) config;
+ this.filePath = buildFilePath();
+ this.fileName = buildFileName();
+ this.flushSize = sinkConfig.getFlushSize();
+ this.hourlyFlushEnabled = sinkConfig.isHourlyFlushEnabled();
+ }
+
+ @Override
+ public void init(ConnectorContext connectorContext) {
+ // init config for hdfs source connector
+ SinkConnectorContext sinkConnectorContext = (SinkConnectorContext)
connectorContext;
+ this.sinkConfig = (FileSinkConfig)
sinkConnectorContext.getSinkConfig();
+ this.fileName = buildFileName();
+ this.filePath = buildFilePath();
+ this.flushSize = sinkConfig.getFlushSize();
+ this.hourlyFlushEnabled = sinkConfig.isHourlyFlushEnabled();
+ }
+
+
+ @Override
+ public void start() throws Exception {
+ if (fileName == null || fileName.length() == 0 || filePath == null ||
filePath.length() == 0) {
+ this.outputStream = System.out;
+ } else {
+ this.outputStream =
+ new PrintStream(Files.newOutputStream(Paths.get(filePath +
fileName), StandardOpenOption.CREATE, StandardOpenOption.APPEND),
+ false, StandardCharsets.UTF_8.name());
+ }
+ }
+
+ @Override
+ public void commit(ConnectRecord record) {
+
+ }
+
+ @Override
+ public String name() {
+ return this.sinkConfig.getConnectorConfig().getConnectorName();
+ }
+
+ @Override
+ public void stop() {
+ outputStream.flush();
+ outputStream.close();
+ }
+
+ @Override
+ public void put(List<ConnectRecord> sinkRecords) {
+ for (ConnectRecord connectRecord : sinkRecords) {
+ // the file data exceed the flushSize create the new file or
+ // hourlyFlushEnabled && time on the hour
+ if (fileSize.get() >= flushSize || (hourlyFlushEnabled &&
LocalDateTime.now().getHour() == 0)) {
+ log.info("flush the file and open");
+ outputStream.flush();
+ outputStream.close();
+ try {
+ fileSize.set(0);
+ this.outputStream = openWithNewFile();
+ } catch (IOException e) {
+ log.error("create file under path {} error", filePath);
+ throw new RuntimeException(e);
+ }
+ }
+ outputStream.println(new String((byte[]) connectRecord.getData(),
StandardCharsets.UTF_8));
+ fileSize.addAndGet(1);
+ }
+ }
+
+ private String buildFilePath() {
+ Calendar calendar = Calendar.getInstance(Locale.CHINA);
+ int year = calendar.get(Calendar.YEAR);
+ int month = calendar.get(Calendar.MONTH) + 1;
+ int day = calendar.get(Calendar.DATE);
+ String filePath = sinkConfig.getConnectorConfig().getTopic()
+ + File.separator + year + File.separator + month + File.separator
+ day + File.separator;
+ File path = new File(filePath);
+ if (!path.exists()) {
+ if (!path.mkdirs()) {
+ log.error("make file dir {} error", filePath);
+ }
+ }
+ return filePath;
+ }
+
+ private String buildFileName() {
+ Calendar calendar = Calendar.getInstance(Locale.CHINA);
+ long currentTime = calendar.getTime().getTime();
+ return sinkConfig.getConnectorConfig().getTopic() + "-" +
calendar.get(Calendar.HOUR_OF_DAY) + "-" + currentTime;
+ }
+
+ private PrintStream openWithNewFile() throws IOException {
+ this.filePath = buildFilePath();
+ this.fileName = buildFileName();
+ if (fileName.length() == 0 || filePath == null || filePath.length() ==
0) {
+ return System.out;
+ }
+ return new PrintStream(Files.newOutputStream(Paths.get(filePath +
fileName),
+ StandardOpenOption.CREATE, StandardOpenOption.APPEND),
+ false, StandardCharsets.UTF_8.name());
+ }
+}
diff --git
a/eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/source/config/FileSourceConfig.java
b/eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/source/config/FileSourceConfig.java
new file mode 100644
index 000000000..b96954440
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/source/config/FileSourceConfig.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.file.source.config;
+
+import org.apache.eventmesh.openconnect.api.config.SourceConfig;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class FileSourceConfig extends SourceConfig {
+
+ public SourceConnectorConfig connectorConfig;
+}
diff --git
a/eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/source/config/SourceConnectorConfig.java
b/eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/source/config/SourceConnectorConfig.java
new file mode 100644
index 000000000..d92b7e04d
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/source/config/SourceConnectorConfig.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.eventmesh.connector.file.source.config;
+
+import lombok.Data;
+
+@Data
+public class SourceConnectorConfig {
+
+ private String connectorName;
+
+ private String nameserver;
+
+ private String topic;
+
+ private long commitOffsetIntervalMs = 5 * 1000;
+}
diff --git
a/eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/source/connector/FileSourceConnector.java
b/eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/source/connector/FileSourceConnector.java
new file mode 100644
index 000000000..1042bc0a2
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-file/src/main/java/org/apache/eventmesh/connector/file/source/connector/FileSourceConnector.java
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.file.source.connector;
+
+import org.apache.eventmesh.connector.file.source.config.FileSourceConfig;
+import org.apache.eventmesh.openconnect.api.config.Config;
+import org.apache.eventmesh.openconnect.api.connector.ConnectorContext;
+import org.apache.eventmesh.openconnect.api.connector.SourceConnectorContext;
+import org.apache.eventmesh.openconnect.api.source.Source;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+import
org.apache.eventmesh.openconnect.offsetmgmt.api.storage.OffsetStorageReader;
+
+
+import java.util.List;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class FileSourceConnector implements Source {
+
+ private FileSourceConfig sourceConfig;
+
+ private OffsetStorageReader offsetStorageReader;
+
+
+ @Override
+ public Class<? extends Config> configClass() {
+ return FileSourceConfig.class;
+ }
+
+ @Override
+ public void init(Config config) throws Exception {
+ // init config for hdfs source connector
+ this.sourceConfig = (FileSourceConfig) config;
+
+ }
+
+ @Override
+ public void init(ConnectorContext connectorContext) throws Exception {
+ SourceConnectorContext sourceConnectorContext =
(SourceConnectorContext) connectorContext;
+ this.sourceConfig = (FileSourceConfig)
sourceConnectorContext.getSourceConfig();
+ this.offsetStorageReader =
sourceConnectorContext.getOffsetStorageReader();
+
+ }
+
+ @Override
+ public void start() throws Exception {
+
+ }
+
+
+ @Override
+ public void commit(ConnectRecord record) {
+
+ }
+
+ @Override
+ public String name() {
+ return this.sourceConfig.getConnectorConfig().getConnectorName();
+ }
+
+ @Override
+ public void stop() {
+
+ }
+
+ @Override
+ public List<ConnectRecord> poll() {
+ return null;
+ }
+
+}
diff --git
a/eventmesh-connectors/eventmesh-connector-file/src/main/resources/server-config.yml
b/eventmesh-connectors/eventmesh-connector-file/src/main/resources/server-config.yml
new file mode 100644
index 000000000..d9416e1ed
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-file/src/main/resources/server-config.yml
@@ -0,0 +1,19 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+sourceEnable: false
+sinkEnable: true
diff --git
a/eventmesh-connectors/eventmesh-connector-file/src/main/resources/sink-config.yml
b/eventmesh-connectors/eventmesh-connector-file/src/main/resources/sink-config.yml
new file mode 100644
index 000000000..03789687e
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-file/src/main/resources/sink-config.yml
@@ -0,0 +1,29 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+pubSubConfig:
+ meshAddress: 127.0.0.1:10000
+ subject: TopicTest
+ idc: FT
+ env: PRD
+ group: fileSink
+ appId: 5031
+ userName: fileSinkUser
+ passWord: filePassWord
+connectorConfig:
+ connectorName: fileSink
+ topic: TopicTest
diff --git
a/eventmesh-connectors/eventmesh-connector-file/src/main/resources/source-config.yml
b/eventmesh-connectors/eventmesh-connector-file/src/main/resources/source-config.yml
new file mode 100644
index 000000000..f1d1efc42
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-file/src/main/resources/source-config.yml
@@ -0,0 +1,40 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+pubSubConfig:
+ meshAddress: 127.0.0.1:10000
+ subject: TopicTest
+ idc: FT
+ env: PRD
+ group: fileSource
+ appId: 5032
+ userName: fileSourceUser
+ passWord: filePassWord
+connectorConfig:
+ connectorName: fileSource
+ nameserver: 127.0.0.1:9877
+ topic: TopicTest
+ commitOffsetIntervalMs: 5000
+offsetStorageConfig:
+ offsetStorageType: nacos
+ offsetStorageAddr: 127.0.0.1:8848
+ extensions: {
+ #same with topic
+ dataId: TopicTest,
+ #same with group
+ group: fileSource
+ }
diff --git a/settings.gradle b/settings.gradle
index ac21f39ff..bdb43facf 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -39,6 +39,7 @@ include 'eventmesh-connectors:eventmesh-connector-s3'
include 'eventmesh-connectors:eventmesh-connector-pravega'
include 'eventmesh-connectors:eventmesh-connector-knative'
include 'eventmesh-connectors:eventmesh-connector-jdbc'
+include 'eventmesh-connectors:eventmesh-connector-file'
include 'eventmesh-storage-plugin:eventmesh-storage-api'
include 'eventmesh-storage-plugin:eventmesh-storage-standalone'
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
