This is an automated email from the ASF dual-hosted git repository.
mikexue 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 c29ada62f [ISSUE #4082][Task1] Support S3 file source connector (#4132)
c29ada62f is described below
commit c29ada62fcdd12750b683b65b57079eb5316ac53
Author: TheR1sing3un <[email protected]>
AuthorDate: Tue Aug 15 09:44:47 2023 +0800
[ISSUE #4082][Task1] Support S3 file source connector (#4132)
* feat(connector): add S3 file source connector with specified format
1. add S3 file source connector with specified format
Closes https://github.com/apache/eventmesh/issues/4082
* test(connector): add S3SourceConnectorTest to verify
1. add S3SourceConnectorTest to verify
Closes https://github.com/apache/eventmesh/issues/4082
* feat(connector): add S3SourceWorker
1. add S3SourceWorker
Closes https://github.com/apache/eventmesh/issues/4082
* style(connector): add header license
1. add header license
Closes https://github.com/apache/eventmesh/issues/4082
* refactor(connector): refactor S3 source connector
1. refactor S3 source connector
Closes https://github.com/apache/eventmesh/issues/4082
---
build.gradle | 2 +
.../eventmesh-connector-s3/build.gradle | 23 +++
.../eventmesh-connector-s3/gradle.properties | 16 ++
.../connector/s3/config/S3ServerConfig.java | 33 +++++
.../connector/s3/server/S3ConnectServer.java | 41 ++++++
.../connector/s3/source/config/S3SourceConfig.java | 29 ++++
.../s3/source/config/SourceConnectorConfig.java | 54 +++++++
.../s3/source/connector/S3SourceConnector.java | 161 +++++++++++++++++++++
.../connector/s3/source/S3SourceConnectorTest.java | 136 +++++++++++++++++
settings.gradle | 1 +
10 files changed, 496 insertions(+)
diff --git a/build.gradle b/build.gradle
index a30f89c3b..e368eb58c 100644
--- a/build.gradle
+++ b/build.gradle
@@ -527,6 +527,8 @@ subprojects {
dependency "com.github.seancfoley:ipaddress:5.3.3"
dependency "javax.annotation:javax.annotation-api:1.3.2"
dependency "com.alibaba:fastjson:1.2.83"
+
+ dependency "software.amazon.awssdk:s3:2.20.29"
}
}
}
diff --git a/eventmesh-connectors/eventmesh-connector-s3/build.gradle
b/eventmesh-connectors/eventmesh-connector-s3/build.gradle
new file mode 100644
index 000000000..9d004b2f1
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-s3/build.gradle
@@ -0,0 +1,23 @@
+/*
+ * 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 {
+ implementation project(":eventmesh-openconnect:eventmesh-openconnect-java")
+ implementation 'software.amazon.awssdk:s3'
+ compileOnly 'org.projectlombok:lombok'
+ annotationProcessor 'org.projectlombok:lombok'
+}
\ No newline at end of file
diff --git a/eventmesh-connectors/eventmesh-connector-s3/gradle.properties
b/eventmesh-connectors/eventmesh-connector-s3/gradle.properties
new file mode 100644
index 000000000..a9fd83fea
--- /dev/null
+++ b/eventmesh-connectors/eventmesh-connector-s3/gradle.properties
@@ -0,0 +1,16 @@
+#
+# 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.
+#
\ No newline at end of file
diff --git
a/eventmesh-connectors/eventmesh-connector-s3/src/main/java/org/apache/eventmesh/connector/s3/config/S3ServerConfig.java
b/eventmesh-connectors/eventmesh-connector-s3/src/main/java/org/apache/eventmesh/connector/s3/config/S3ServerConfig.java
new file mode 100644
index 000000000..f5c4371d4
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-s3/src/main/java/org/apache/eventmesh/connector/s3/config/S3ServerConfig.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.s3.config;
+
+import org.apache.eventmesh.openconnect.api.config.Config;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class S3ServerConfig extends Config {
+
+ private boolean sourceEnable;
+
+ private boolean sinkEnable;
+
+}
diff --git
a/eventmesh-connectors/eventmesh-connector-s3/src/main/java/org/apache/eventmesh/connector/s3/server/S3ConnectServer.java
b/eventmesh-connectors/eventmesh-connector-s3/src/main/java/org/apache/eventmesh/connector/s3/server/S3ConnectServer.java
new file mode 100644
index 000000000..e1b485801
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-s3/src/main/java/org/apache/eventmesh/connector/s3/server/S3ConnectServer.java
@@ -0,0 +1,41 @@
+/*
+ * 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.s3.server;
+
+import org.apache.eventmesh.connector.s3.config.S3ServerConfig;
+import org.apache.eventmesh.connector.s3.source.connector.S3SourceConnector;
+import org.apache.eventmesh.openconnect.Application;
+import org.apache.eventmesh.openconnect.util.ConfigUtil;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class S3ConnectServer {
+
+ public static void main(String[] args) throws Exception {
+ S3ServerConfig s3ServerConfig = ConfigUtil.parse(S3ServerConfig.class,
"server-config.yml");
+ if (s3ServerConfig.isSourceEnable()) {
+ Application application = new Application();
+ application.run(S3SourceConnector.class);
+ }
+
+ if (s3ServerConfig.isSinkEnable()) {
+ log.error("S3 sink is not supported yet.");
+ }
+ }
+}
diff --git
a/eventmesh-connectors/eventmesh-connector-s3/src/main/java/org/apache/eventmesh/connector/s3/source/config/S3SourceConfig.java
b/eventmesh-connectors/eventmesh-connector-s3/src/main/java/org/apache/eventmesh/connector/s3/source/config/S3SourceConfig.java
new file mode 100644
index 000000000..3355e8438
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-s3/src/main/java/org/apache/eventmesh/connector/s3/source/config/S3SourceConfig.java
@@ -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.
+ */
+
+
+package org.apache.eventmesh.connector.s3.source.config;
+
+import org.apache.eventmesh.openconnect.api.config.SourceConfig;
+
+import lombok.Data;
+
+
+@Data
+public class S3SourceConfig extends SourceConfig {
+ private SourceConnectorConfig sourceConnectorConfig;
+}
diff --git
a/eventmesh-connectors/eventmesh-connector-s3/src/main/java/org/apache/eventmesh/connector/s3/source/config/SourceConnectorConfig.java
b/eventmesh-connectors/eventmesh-connector-s3/src/main/java/org/apache/eventmesh/connector/s3/source/config/SourceConnectorConfig.java
new file mode 100644
index 000000000..314ae4b1f
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-s3/src/main/java/org/apache/eventmesh/connector/s3/source/config/SourceConnectorConfig.java
@@ -0,0 +1,54 @@
+/*
+ * 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.s3.source.config;
+
+import java.util.Map;
+
+import lombok.Data;
+
+@Data
+public class SourceConnectorConfig {
+
+ private String connectorName;
+
+ private String region;
+
+ private String bucket;
+
+ private String accessKey;
+
+ private String secretKey;
+
+ private String fileName;
+
+ /**
+ * The schema for the data to be read from S3.
+ */
+ private Map<String/*column name*/, Integer/*bytes*/> schema;
+
+ /**
+ * The maximum number of records that should be returned in each batch
poll.
+ */
+ private Integer batchSize = 20;
+
+ /**
+ * The maximum ms to wait for request futures to complete.
+ */
+ private long timeout = 3000;
+}
diff --git
a/eventmesh-connectors/eventmesh-connector-s3/src/main/java/org/apache/eventmesh/connector/s3/source/connector/S3SourceConnector.java
b/eventmesh-connectors/eventmesh-connector-s3/src/main/java/org/apache/eventmesh/connector/s3/source/connector/S3SourceConnector.java
new file mode 100644
index 000000000..43b757d08
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-s3/src/main/java/org/apache/eventmesh/connector/s3/source/connector/S3SourceConnector.java
@@ -0,0 +1,161 @@
+/*
+ * 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.s3.source.connector;
+
+
+import org.apache.eventmesh.connector.s3.source.config.S3SourceConfig;
+import org.apache.eventmesh.connector.s3.source.config.SourceConnectorConfig;
+import org.apache.eventmesh.openconnect.api.config.Config;
+import org.apache.eventmesh.openconnect.api.source.Source;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordOffset;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordPartition;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import lombok.extern.slf4j.Slf4j;
+
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.core.ResponseBytes;
+import software.amazon.awssdk.core.async.AsyncResponseTransformer;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
+
+@Slf4j
+public class S3SourceConnector implements Source {
+
+ public static final String REGION = "region";
+
+ public static final String BUCKET = "bucket";
+
+ public static final String FILE_NAME = "fileName";
+
+ public static final String POSITION = "position";
+
+ private S3SourceConfig sourceConfig;
+
+ private SourceConnectorConfig sourceConnectorConfig;
+
+ private int eachRecordSize;
+
+ private long fileSize;
+
+ private S3AsyncClient s3Client;
+
+ private long position;
+
+
+ @Override
+ public Class<? extends Config> configClass() {
+ return S3SourceConfig.class;
+ }
+
+ @Override
+ public void init(Config config) throws Exception {
+ // init config for s3 source connector
+ this.sourceConfig = (S3SourceConfig) config;
+ this.sourceConnectorConfig =
this.sourceConfig.getSourceConnectorConfig();
+ this.eachRecordSize = calculateEachRecordSize();
+ AwsBasicCredentials basicCredentials =
AwsBasicCredentials.create(this.sourceConnectorConfig.getAccessKey(),
+ this.sourceConnectorConfig.getSecretKey());
+ this.s3Client = S3AsyncClient.builder().credentialsProvider(() ->
basicCredentials)
+ .region(Region.of(this.sourceConnectorConfig.getRegion())).build();
+ }
+
+ private int calculateEachRecordSize() {
+ Optional<Integer> sum =
this.sourceConnectorConfig.getSchema().values().stream().reduce((x, y) -> x +
y);
+ return sum.orElse(0);
+ }
+
+ @Override
+ public void start() throws Exception {
+ CompletableFuture<HeadObjectResponse>
headObjectResponseCompletableFuture = this.s3Client.headObject(
+ builder ->
builder.bucket(this.sourceConnectorConfig.getBucket()).key(this.sourceConnectorConfig.getFileName()));
+
headObjectResponseCompletableFuture.get(this.sourceConnectorConfig.getTimeout(),
TimeUnit.MILLISECONDS);
+ this.fileSize =
headObjectResponseCompletableFuture.get().contentLength();
+ }
+
+ @Override
+ public void commit(ConnectRecord record) {
+
+ }
+
+ @Override
+ public String name() {
+ return this.sourceConfig.getSourceConnectorConfig().getConnectorName();
+ }
+
+ @Override
+ public void stop() throws Exception {
+
+ }
+
+ @Override
+ public List<ConnectRecord> poll() {
+ if (this.position >= this.fileSize) {
+ return Collections.EMPTY_LIST;
+ }
+ long startPosition = this.position;
+ long endPosition = Math.min(this.fileSize, this.position +
this.eachRecordSize * this.sourceConnectorConfig.getBatchSize()) - 1;
+ GetObjectRequest request =
GetObjectRequest.builder().bucket(this.sourceConnectorConfig.getBucket())
+ .key(this.sourceConnectorConfig.getFileName())
+ .range("bytes=" + startPosition + "-" + endPosition).build();
+ ResponseBytes<GetObjectResponse> resp;
+ try {
+ resp = this.s3Client.getObject(request,
AsyncResponseTransformer.toBytes())
+ .get(this.sourceConnectorConfig.getTimeout(),
TimeUnit.MILLISECONDS);
+ } catch (Exception e) {
+ log.error("poll records from S3 file, poll range {}-{}, but
failed", startPosition, endPosition, e);
+ return Collections.EMPTY_LIST;
+ }
+ byte[] bytes = resp.asByteArray();
+ List<ConnectRecord> records = new ArrayList<>(bytes.length /
this.eachRecordSize);
+ for (int i = 0; i < bytes.length; i += this.eachRecordSize) {
+ byte[] body = new byte[this.eachRecordSize];
+ System.arraycopy(bytes, i, body, 0, this.eachRecordSize);
+ this.position += this.eachRecordSize;
+ ConnectRecord record = new ConnectRecord(getRecordPartition(),
getRecordOffset(), System.currentTimeMillis(), body);
+ records.add(record);
+ }
+ return records;
+ }
+
+ private RecordPartition getRecordPartition() {
+ Map<String, String> map = new HashMap<>();
+ map.put(REGION, this.sourceConnectorConfig.getRegion());
+ map.put(BUCKET, this.sourceConnectorConfig.getBucket());
+ map.put(FILE_NAME, this.sourceConnectorConfig.getFileName());
+ return new RecordPartition(map);
+ }
+
+ private RecordOffset getRecordOffset() {
+ Map<String, String> map = new HashMap<>();
+ map.put(POSITION, String.valueOf(this.position));
+ return new RecordOffset(map);
+ }
+}
diff --git
a/eventmesh-connectors/eventmesh-connector-s3/src/test/java/org/apache/eventmesh/connector/s3/source/S3SourceConnectorTest.java
b/eventmesh-connectors/eventmesh-connector-s3/src/test/java/org/apache/eventmesh/connector/s3/source/S3SourceConnectorTest.java
new file mode 100644
index 000000000..27ab9d96e
--- /dev/null
+++
b/eventmesh-connectors/eventmesh-connector-s3/src/test/java/org/apache/eventmesh/connector/s3/source/S3SourceConnectorTest.java
@@ -0,0 +1,136 @@
+/*
+ * 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.s3.source;
+
+import org.apache.eventmesh.connector.s3.source.config.S3SourceConfig;
+import org.apache.eventmesh.connector.s3.source.config.SourceConnectorConfig;
+import org.apache.eventmesh.connector.s3.source.connector.S3SourceConnector;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.core.async.AsyncRequestBody;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.model.PutObjectRequest;
+
+
+@Ignore
+public class S3SourceConnectorTest {
+
+ private static final S3SourceConfig sourceConfig;
+
+ private static final SourceConnectorConfig SOURCE_CONNECTOR_CONFIG;
+
+ private static final Map<String, Integer> schema;
+
+ private static final int eachRecordSize;
+
+ static {
+ sourceConfig = new S3SourceConfig();
+ SOURCE_CONNECTOR_CONFIG = new SourceConnectorConfig();
+ SOURCE_CONNECTOR_CONFIG.setConnectorName("S3SourceConnector");
+ SOURCE_CONNECTOR_CONFIG.setRegion("ap-southeast-1");
+ SOURCE_CONNECTOR_CONFIG.setBucket("event-mesh-bucket");
+ SOURCE_CONNECTOR_CONFIG.setAccessKey("access-key");
+ SOURCE_CONNECTOR_CONFIG.setSecretKey("secret-key");
+
+ SOURCE_CONNECTOR_CONFIG.setFileName("test-file");
+
+ schema = new HashMap<>();
+ schema.put("id", 4);
+ schema.put("body", 16);
+ schema.put("time", 8);
+
+ eachRecordSize = schema.values().stream().reduce((x, y) -> x +
y).orElse(0);
+ SOURCE_CONNECTOR_CONFIG.setSchema(schema);
+ sourceConfig.setSourceConnectorConfig(SOURCE_CONNECTOR_CONFIG);
+ }
+
+ private S3AsyncClient s3Client;
+
+ @Before
+ public void setUp() throws Exception {
+ AwsBasicCredentials basicCredentials =
AwsBasicCredentials.create(this.SOURCE_CONNECTOR_CONFIG.getAccessKey(),
+ this.SOURCE_CONNECTOR_CONFIG.getSecretKey());
+ this.s3Client = S3AsyncClient.builder().credentialsProvider(() ->
basicCredentials)
+
.region(Region.of(this.SOURCE_CONNECTOR_CONFIG.getRegion())).build();
+
+ // write mocked data
+ this.writeMockedRecords(200);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ // clear file
+ this.s3Client.deleteObject(builder ->
builder.bucket(this.SOURCE_CONNECTOR_CONFIG.getBucket())
+ .key(this.SOURCE_CONNECTOR_CONFIG.getFileName()).build()).get();
+ }
+
+ @Test
+ public void testS3SourceConnector() throws Exception {
+ S3SourceConnector s3SourceConnector = new S3SourceConnector();
+ s3SourceConnector.init(sourceConfig);
+ s3SourceConnector.start();
+ int expectedId = 0;
+ while (true) {
+ List<ConnectRecord> connectRecords = s3SourceConnector.poll();
+ if (connectRecords.isEmpty()) {
+ break;
+ }
+ Assert.assertEquals(20, connectRecords.size());
+ for (ConnectRecord connectRecord : connectRecords) {
+ byte[] data = (byte[]) connectRecord.getData();
+ Assert.assertEquals(eachRecordSize, data.length);
+ ByteBuffer byteBuffer = ByteBuffer.wrap(data);
+ int id = byteBuffer.getInt();
+ Assert.assertEquals(expectedId++, id);
+ }
+ }
+
+ }
+
+ private void writeMockedRecords(int count) throws Exception {
+ ByteBuffer bytes = ByteBuffer.allocate(count * eachRecordSize);
+ ByteBuffer body = ByteBuffer.allocate(16);
+ body.putLong(13L);
+ body.putLong(13L);
+ body.flip();
+ for (int i = 0; i < count; i++) {
+ bytes.putInt(i);
+ bytes.put(body);
+ body.flip();
+ bytes.putLong(System.currentTimeMillis());
+ }
+ PutObjectRequest putObjectRequest = PutObjectRequest.builder()
+
.bucket(SOURCE_CONNECTOR_CONFIG.getBucket()).key(SOURCE_CONNECTOR_CONFIG.getFileName()).build();
+ AsyncRequestBody requestBody =
AsyncRequestBody.fromBytes(bytes.array());
+ this.s3Client.putObject(putObjectRequest, requestBody).get();
+ }
+
+}
diff --git a/settings.gradle b/settings.gradle
index fe42d4030..588cbddc5 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -81,3 +81,4 @@ include 'eventmesh-webhook:eventmesh-webhook-receive'
include 'eventmesh-security-plugin'
include 'eventmesh-security-plugin:eventmesh-security-api'
include 'eventmesh-security-plugin:eventmesh-security-auth-token'
+include 'eventmesh-connectors:eventmesh-connector-s3'
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
