[ISSUE #4635] Implemented the functions of file source connector #4650

HarshSawarkar · 2023-12-14T10:09:42Z

Fixes #4635 .

Motivation

File source connector class methods were not implemented.

Modifications

Successfully implemented start(), stop(), poll() methods of file source connector class.

Documentation

Does this pull request introduce a new feature? (no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
If a feature is not applicable for documentation, explain why?
If a feature is not documented yet in this PR, please create a followup issue for adding the documentation

.../src/main/java/org/apache/eventmesh/connector/file/source/connector/FileSourceConnector.java

pandaapo · 2023-12-15T04:28:46Z

...r-file/src/main/java/org/apache/eventmesh/connector/file/source/config/FileSourceConfig.java

 @EqualsAndHashCode(callSuper = true)
 public class FileSourceConfig extends SourceConfig {

    public SourceConnectorConfig connectorConfig;
+    private String fileName;
+    private String filePath;


Can fileName and filePath be directly obtained from the configuration file source-config.yml? Then, it will not restrict users from using specific file name and file path.

In addition, many configuration items in the configuration file source-config.yml are not used by you. Could you directly remove the configuration items that you do not need?

pandaapo · 2023-12-15T04:30:58Z

.../src/main/java/org/apache/eventmesh/connector/file/source/connector/FileSourceConnector.java

@@ -36,6 +47,11 @@ public class FileSourceConnector implements Source {

    private OffsetStorageReader offsetStorageReader;

+    private String filePath;
+    private String fileName;
+    private InputStream inputStream;


After having BufferedReader variable, it seems unnecessary to declare InputStream variable now.

codecov · 2023-12-15T08:35:43Z

Codecov Report

All modified and coverable lines are covered by tests ✅

Comparison is base (b85a58c) 17.39% compared to head (9f1c144) 17.59%.
Report is 36 commits behind head on master.

❗ Current head 9f1c144 differs from pull request most recent head 3729fdb. Consider uploading reports for the commit 3729fdb to get more accurate results

Additional details and impacted files

@@             Coverage Diff              @@
##             master    #4650      +/-   ##
============================================
+ Coverage     17.39%   17.59%   +0.20%     
- Complexity     1757     1774      +17     
============================================
  Files           797      797              
  Lines         29774    29786      +12     
  Branches       2573     2573              
============================================
+ Hits           5178     5242      +64     
+ Misses        24118    24063      -55     
- Partials        478      481       +3

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

Pil0tXia · 2024-01-02T15:09:43Z

eventmesh-connectors/eventmesh-connector-file/src/main/resources/source-config.yml

+    fileName: TopicTest-%s
+    filePath: TopicTest/%s


LGTM. May you please provide a config document here later? That would be helpful.

pandaapo

I would appreciate it if you could test the functionality of this File Source Connector locally. I believe it is a very important job.
For a more visual representation of the test results, you can integrate testing with the File Sink Connector to see if your File Source Connector can transfer the content of one file to another.

pandaapo · 2024-01-03T09:26:44Z

...r-file/src/main/java/org/apache/eventmesh/connector/file/source/config/FileSourceConfig.java


 @Data
+@Slf4j
 @EqualsAndHashCode(callSuper = true)


Where do these two annotations work?

pandaapo · 2024-01-03T09:28:53Z

.../src/main/java/org/apache/eventmesh/connector/file/source/connector/FileSourceConnector.java

+        try {
+            String line;
+            while ((line = bufferedReader.readLine()) != null) {
+                ConnectRecord connectRecord = new ConnectRecord(new RecordPartition(), new RecordOffset(), System.currentTimeMillis(), line);


The offset recording seems redundant because the Source Connector you've implemented reads the entire contents of a file in a single pass and does not perform incremental reads.

Reading the entire file may potentially impose significant memory pressure. It is necessary to set a memory limit or read the file in segments.

I didn't see that this Connector implements memory limitation and fragmented reads through offset. If it is indeed achieved, could you explain how it was achieved? Thank you.

I have addressed the implementation of fragmented reads through offset in the updated code. This was achieved by reading the file in chunks and recording the offset based on the last position index in the buffer array. Additionally, I ensured that the last position is updated in each iteration to maintain accurate tracking. Thank you for bringing this to my attention.

pandaapo · 2024-01-03T09:31:17Z

eventmesh-connectors/eventmesh-connector-file/src/main/resources/source-config.yml

 connectorConfig:
    connectorName: fileSource
    nameserver: 127.0.0.1:9877
    topic: TopicTest
    commitOffsetIntervalMs: 5000
+    fileName: TopicTest-%s
+    filePath: TopicTest/%s


When users use this File Source Connector, they will have to place local files in directories structured like XXX/2024/01/03. I think it is not user-friendly. Why not directly use the filePath and fileName configured by the user here to create input stream?

pandaapo · 2024-01-03T09:32:33Z

eventmesh-connectors/eventmesh-connector-file/src/main/resources/source-config.yml

 connectorConfig:
    connectorName: fileSource
    nameserver: 127.0.0.1:9877
    topic: TopicTest
    commitOffsetIntervalMs: 5000
+    fileName: TopicTest-%s
+    filePath: TopicTest/%s
 offsetStorageConfig:
    offsetStorageType: nacos
    offsetStorageAddr: 127.0.0.1:8848


Regarding these offset configurations, as I mentioned before, the File Source Connector you implemented reads all the content of the file at once, and offsets seem to have no effect. Could offsetStorageConfig and commitOffsetIntervalMs be removed?

I have implemented the commitOffsetIntervalMs in the updated code.

pandaapo · 2024-01-03T09:33:09Z

eventmesh-connectors/eventmesh-connector-file/src/main/resources/source-config.yml

@@ -15,20 +15,13 @@
 # limitations under the License.
 #

-pubSubConfig:


Why was pubSubConfig configuration removed? With this configuration removed, this Source Connector will no longer work properly.

eventmesh-connectors/eventmesh-connector-file/src/main/resources/source-config.yml

pandaapo · 2024-01-09T09:41:20Z

.../src/main/java/org/apache/eventmesh/connector/file/source/connector/FileSourceConnector.java


 import lombok.extern.slf4j.Slf4j;

 @Slf4j
 public class FileSourceConnector implements Source {
+    private static final int DEFAULT_BATCH_SIZE = 10;


What is the role of this variable?

It was used in the initialization of List of ConnectRecord in here

It can be removed.

pandaapo · 2024-01-09T09:41:37Z

.../src/main/java/org/apache/eventmesh/connector/file/source/connector/FileSourceConnector.java

+                ConnectRecord connectRecord = new ConnectRecord(recordPartition, recordOffset, timeStamp, line);
+                connectRecords.add(connectRecord);
+                if (timeStamp - prevTimeStamp > this.sourceConfig.getConnectorConfig().getCommitOffsetIntervalMs()) {
+                    this.commitOffset(connectRecord, lastOffset);


The main purpose of offSet: Source Connector periodically persists the latest offSet somewhere (such as Nacos, Consul, ETCD, etc.). Then, if Source Connector is abruptly interrupted and restarted, it can resume reading from the position indicated by the recorded offSet.
However, for file reading, relying solely on offSet is not feasible because file content does not follow a linear append-only pattern. If file modifications are considered, using offset is more impractical.
Therefore, in this PR, I think the codes related to offSet are not meaningful.

Sure, I will update the PR accordingly.

The main purpose of offSet: Source Connector periodically persists the latest offSet somewhere (such as Nacos, Consul, ETCD, etc.). Then, if Source Connector is abruptly interrupted and restarted, it can resume reading from the position indicated by the recorded offSet. However, for file reading, relying solely on offSet is not feasible because file content does not follow a linear append-only pattern. If file modifications are considered, using offset is more impractical. Therefore, in this PR, I think the codes related to offSet are not meaningful.

@pandaapo Can RecordPartition be considered for recording the last offset position?

Through this explanation, I believe you understand the functionality of offSet now. If you strongly wish to utilize RecordPartition or RecordOffset to implement the functionality of offSet, you can share your proposed approach here first.

Assuming this is the content of a file, and Source Connector starts reading from "A" and stops when it reaches "K".

ABCDEDFG, HIJKLMN, OPQRST.

Your approach needs to consider some scenarios, such as:
(1) After restarting Source Connector, how can it start reading from "L"?

(2) Before Source Connector is restarted, the content of the file is modified as follows. How should it be appropriately handled after restarting Source Connector? read from "L", "A" or other position?

ABCDEDFG 123456, HIJKLMN, OPQRST.

or

ABC, HIJKLMN, OPQRST.

@HarshSawarkar There is no need to consider implementing the concept of offset in the file source connector. For the ability to resume uploading from a breakpoint, you only need to retransmit the entire file content.

pandaapo · 2024-01-13T09:33:03Z

eventmesh-connectors/eventmesh-connector-file/src/main/resources/source-config.yml

@@ -28,13 +27,7 @@ connectorConfig:
    connectorName: fileSource
    nameserver: 127.0.0.1:9877
    topic: TopicTest


nameserver and topic are actually redundant for this Connector.

pandaapo · 2024-01-13T09:33:37Z

.../src/main/java/org/apache/eventmesh/connector/file/source/connector/FileSourceConnector.java

+            while ((bytesRead = bufferedReader.read(buffer)) != -1) {
+                String line = new String(buffer, 0, bytesRead);
+                long timeStamp = System.currentTimeMillis();
+                RecordPartition recordPartition = convertToRecordPartition(this.sourceConfig.getConnectorConfig().getTopic(), fileName);


The contents of this object are the same each time it is created. Either move it outside while () {} or remove it.

pandaapo · 2024-01-13T10:20:32Z

eventmesh-connectors/eventmesh-connector-file/src/main/resources/source-config.yml

@@ -28,13 +27,7 @@ connectorConfig:
    connectorName: fileSource
    nameserver: 127.0.0.1:9877
    topic: TopicTest


nameserver and topic are actually redundant for this Connector.

nameserver and topic are actually redundant for this Connector.

Updated.

pandaapo · 2024-01-13T12:04:34Z

eventmesh-connectors/eventmesh-connector-file/src/main/resources/source-config.yml

@@ -26,15 +25,7 @@ pubSubConfig:
    passWord: filePassWord
 connectorConfig:
    connectorName: fileSource
-    nameserver: 127.0.0.1:9877
    topic: TopicTest


topic is redundant, too.

@pandaapo Sure, I will update the PR. Just for my knowledge, I wanted to know the usage of topic in the source connector. Also, in the ConnectorConfig, the topic was used only once, as in every other source-config.yml file for other source connectors. So, would removing topic be a good idea?

topic is used as part of the data source definition in other Source Connectors, such as fetching data from a topic of Kafka, RocketMQ, or RabbitMQ. The current Connector fetches data from a file, and filePath and fileName already specify the data source.

Updated the PR.Can you please check now?

OK. I am now starting the CI for this PR. I would appreciate it if you could test the functionality of this File Source Connector locally. I believe it is a very important job.

pandaapo · 2024-01-14T09:58:02Z

.../src/main/java/org/apache/eventmesh/connector/file/source/connector/FileSourceConnector.java

+            this.bufferedReader = new BufferedReader(new InputStreamReader(System.in, StandardCharsets.UTF_8));
+        } else {
+            this.bufferedReader = new BufferedReader(new InputStreamReader(Files.newInputStream(
+                Paths.get(filePath, fileName)), StandardCharsets.UTF_8), BUFFER_SIZE);


In your new commit

I think there is no need to make separate judgments for fileName and filePath in this conditional judgment if (fileName == null || fileName.isEmpty() || filePath == null || filePath.isEmpty()).

What will Paths.get("/path/to/example.txt", "example.txt") return?

I would appreciate it if you could test the functionality of this File Source Connector locally. I believe it is a very important job.

Updated.Also I have tested the File Source Connector for start(), stop() and poll() method, locally, and it's running fine.I have attached the Ss of the tests I ran on this class below.

pandaapo

Could you submit your unit test code to this PR as well?

pandaapo · 2024-01-19T12:39:39Z

.../src/main/java/org/apache/eventmesh/connector/file/source/connector/FileSourceConnector.java

-    private OffsetStorageReader offsetStorageReader;
+    private String filePath;
+    private String fileName;
+    public BufferedReader bufferedReader;


Why do you expand the scope of access modifiers for bufferedReader?

pandaapo · 2024-01-19T13:12:38Z

eventmesh-connectors/eventmesh-connector-file/build.gradle

@@ -17,6 +17,8 @@

 dependencies {
    api project(":eventmesh-openconnect:eventmesh-openconnect-java")
+    implementation 'org.junit.jupiter:junit-jupiter:5.8.1'
+    implementation 'org.mockito:mockito-junit-jupiter:5.3.1'


Please change implementation to testImplementation. Also, there is no need to specify the version number of these 2 jars additionally.

pandaapo · 2024-01-19T13:13:01Z

.../main/java/org/apache/eventmesh/connector/file/source/connector/FileSourceConnectorTest.java

+        fileSourceConnector.start();
+        List<ConnectRecord> connectRecords = fileSourceConnector.poll();
+        fileSourceConnector.stop();
+        Assertions.assertEquals(content, connectRecords.get(0).getData().toString());


Is it possible to delete the d/f/foo.txt file created at the end of the test?

pandaapo · 2024-01-19T13:27:53Z

.../main/java/org/apache/eventmesh/connector/file/source/connector/FileSourceConnectorTest.java

+import org.junit.jupiter.api.Test;
+import org.mockito.Mock;
+
+class FileSourceConnectorTest {


Unit test classes should be placed in the 'src/test' directory, not 'src/main' directory.

Implemented the functions of file source connector.

3a2e502

HarshSawarkar mentioned this pull request Dec 14, 2023

[Enhancement] Implement the function of File Source Connector #4635

Closed

3 tasks

Pil0tXia requested changes Dec 14, 2023

View reviewed changes

.../src/main/java/org/apache/eventmesh/connector/file/source/connector/FileSourceConnector.java Outdated Show resolved Hide resolved

pandaapo reviewed Dec 15, 2023

View reviewed changes

xwm1992 changed the title ~~Implemented the functions of file source connector.~~ [ISSUE #4635]Implemented the functions of file source connector. Dec 22, 2023

Implemented the functions of file source connector.

5f5c9c3

HarshSawarkar requested review from pandaapo and Pil0tXia January 2, 2024 13:12

Pil0tXia previously approved these changes Jan 2, 2024

View reviewed changes

pandaapo reviewed Jan 3, 2024

View reviewed changes

HarshSawarkar added 2 commits January 8, 2024 16:30

Implemented the functions of file source connector.

9fbae2d

Implemented the functions of file source connector.

e2de7cd

HarshSawarkar dismissed Pil0tXia’s stale review via e2de7cd January 8, 2024 11:18

HarshSawarkar requested review from pandaapo and Pil0tXia January 8, 2024 11:22

Pil0tXia reviewed Jan 9, 2024

View reviewed changes

eventmesh-connectors/eventmesh-connector-file/src/main/resources/source-config.yml Outdated Show resolved Hide resolved

pandaapo reviewed Jan 9, 2024

View reviewed changes

Implemented the functions of file source connector.

43bfee2

HarshSawarkar requested review from xwm1992, Pil0tXia and pandaapo January 13, 2024 06:40

pandaapo reviewed Jan 13, 2024

View reviewed changes

Implemented the functions of file source connector.

e5ae581

HarshSawarkar requested a review from pandaapo January 13, 2024 09:51

pandaapo reviewed Jan 13, 2024

View reviewed changes

Implemented the functions of file source connector.

206b08c

HarshSawarkar requested a review from pandaapo January 13, 2024 11:43

pandaapo reviewed Jan 13, 2024

View reviewed changes

Implemented the functions of file source connector.

60ec82d

HarshSawarkar requested a review from pandaapo January 13, 2024 13:04

Implemented the functions of file source connector.

9f1c144

pandaapo reviewed Jan 14, 2024

View reviewed changes

Implemented the functions of file source connector.

7c4c051

pandaapo reviewed Jan 19, 2024

View reviewed changes

Implemented the functions of file source connector.

c8a45f6

pandaapo reviewed Jan 19, 2024

View reviewed changes

Implemented the functions of file source connector.

3729fdb

pandaapo approved these changes Jan 20, 2024

View reviewed changes

Pil0tXia approved these changes Jan 20, 2024

View reviewed changes

Pil0tXia changed the title ~~[ISSUE #4635]Implemented the functions of file source connector.~~ [ISSUE #4635] Implemented the functions of file source connector Jan 20, 2024

Pil0tXia merged commit 21e4c3a into apache:master Jan 20, 2024
9 checks passed

[ISSUE #4635] Implemented the functions of file source connector #4650

[ISSUE #4635] Implemented the functions of file source connector #4650

Conversation

HarshSawarkar commented Dec 14, 2023

Motivation

Modifications

Documentation

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov bot commented Dec 15, 2023 • edited Loading

Codecov Report

Pil0tXia Jan 2, 2024 • edited Loading

Choose a reason for hiding this comment

pandaapo left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pandaapo Jan 3, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pandaapo Jan 3, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

HarshSawarkar Jan 19, 2024 • edited Loading

Choose a reason for hiding this comment

pandaapo left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov bot commented Dec 15, 2023 •

edited

Loading

Pil0tXia Jan 2, 2024 •

edited

Loading

pandaapo Jan 3, 2024 •

edited

Loading

pandaapo Jan 3, 2024 •

edited

Loading

HarshSawarkar Jan 19, 2024 •

edited

Loading