Skip to content

Commit b7002bf

Browse files
ic4yHisoka-X
ic4y
and
Hisoka-X
authored
[Feature][Connector-V2] Add web3j source connector (#6598)
 --------- Co-authored-by: Jia Fan <[email protected]>
1 parent fc39390 commit b7002bf

File tree

15 files changed

+566
-0
lines changed

15 files changed

+566
-0
lines changed

‎docs/en/connector-v2/source/Web3j.md

+61
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
# Web3j
2+
3+
> Web3j source connector
4+
5+
## Support Those Engines
6+
7+
> Spark<br/>
8+
> Flink<br/>
9+
> Seatunnel Zeta<br/>
10+
11+
## Key Features
12+
13+
- [x] [batch](../../concept/connector-v2-features.md)
14+
- [x] [stream](../../concept/connector-v2-features.md)
15+
- [ ] [exactly-once](../../concept/connector-v2-features.md)
16+
- [ ] [column projection](../../concept/connector-v2-features.md)
17+
- [ ] [parallelism](../../concept/connector-v2-features.md)
18+
- [ ] [support user-defined split](../../concept/connector-v2-features.md)
19+
20+
## Description
21+
22+
Source connector for web3j. It is used to read data from the blockchain, such as block information, transactions, smart contract events, etc. Currently, it supports reading block height data.
23+
24+
## Source Options
25+
26+
| Name | Type | Required | Default | Description |
27+
|------|--------|----------|---------|---------------------------------------------------------------------------------------------------------|
28+
| url | String | Yes | - | When using Infura as the service provider, the URL is used for communication with the Ethereum network. |
29+
30+
## How to Create a Http Data Synchronization Jobs
31+
32+
```hocon
33+
env {
34+
parallelism = 1
35+
job.mode = "BATCH"
36+
}
37+
38+
source {
39+
Web3j {
40+
url = "https://mainnet.infura.io/v3/xxxxx"
41+
}
42+
}
43+
44+
# Console printing of the read Http data
45+
sink {
46+
Console {
47+
parallelism = 1
48+
}
49+
}
50+
```
51+
52+
Then you will get the following data:
53+
54+
```json
55+
{"blockNumber":19525949,"timestamp":"2024-03-27T13:28:45.605Z"}
56+
```
57+
58+
## Changelog
59+
60+
- Add Web3j Source Connector
61+

‎plugin-mapping.properties

+1
Original file line numberDiff line numberDiff line change
@@ -105,6 +105,7 @@ seatunnel.sink.Maxcompute = connector-maxcompute
105105
seatunnel.source.MySQL-CDC = connector-cdc-mysql
106106
seatunnel.source.MongoDB-CDC = connector-cdc-mongodb
107107
seatunnel.sink.S3Redshift = connector-s3-redshift
108+
seatunnel.source.Web3j = connector-web3j
108109
seatunnel.source.TDengine = connector-tdengine
109110
seatunnel.sink.TDengine = connector-tdengine
110111
seatunnel.source.Persistiq = connector-http-persistiq

‎release-note.md

+1
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
- [Hbase] Add hbase sink connector #4049
1919
- [Clickhouse] Fix clickhouse old version compatibility #5326
2020
- [Easysearch] Support INFINI Easysearch #5933
21+
- [Web3j] add Web3j source connector #6598
2122
### Formats
2223
- [Canal]Support read canal format message #3950
2324
- [Debezium]Support debezium canal format message #3981

‎seatunnel-connectors-v2/connector-web3j/pom.xml

+54
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<!--
3+
Licensed to the Apache Software Foundation (ASF) under one or more
4+
contributor license agreements. See the NOTICE file distributed with
5+
this work for additional information regarding copyright ownership.
6+
The ASF licenses this file to You under the Apache License, Version 2.0
7+
(the "License"); you may not use this file except in compliance with
8+
the License. You may obtain a copy of the License at
9+
10+
http://www.apache.org/licenses/LICENSE-2.0
11+
12+
Unless required by applicable law or agreed to in writing, software
13+
distributed under the License is distributed on an "AS IS" BASIS,
14+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
See the License for the specific language governing permissions and
16+
limitations under the License.
17+
-->
18+
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
19+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
20+
<modelVersion>4.0.0</modelVersion>
21+
<parent>
22+
<groupId>org.apache.seatunnel</groupId>
23+
<artifactId>seatunnel-connectors-v2</artifactId>
24+
<version>${revision}</version>
25+
</parent>
26+
27+
<artifactId>connector-web3j</artifactId>
28+
<name>SeaTunnel : Connectors V2 : Web3j</name>
29+
30+
<properties>
31+
<web3j.version>4.8.4</web3j.version>
32+
</properties>
33+
34+
<dependencies>
35+
<dependency>
36+
<groupId>org.apache.seatunnel</groupId>
37+
<artifactId>connector-common</artifactId>
38+
<version>${project.version}</version>
39+
</dependency>
40+
41+
<dependency>
42+
<groupId>org.apache.seatunnel</groupId>
43+
<artifactId>seatunnel-format-json</artifactId>
44+
<version>${project.version}</version>
45+
</dependency>
46+
47+
<dependency>
48+
<groupId>org.web3j</groupId>
49+
<artifactId>core</artifactId>
50+
<version>${web3j.version}</version>
51+
</dependency>
52+
</dependencies>
53+
54+
</project>

‎seatunnel-connectors-v2/connector-web3j/src/main/java/org/apache/seatunnel/connectors/seatunnel/config/Web3jConfig.java

+31
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.seatunnel.connectors.seatunnel.config;
19+
20+
import org.apache.seatunnel.api.configuration.Option;
21+
import org.apache.seatunnel.api.configuration.Options;
22+
23+
public class Web3jConfig {
24+
25+
public static final Option<String> URL =
26+
Options.key("url")
27+
.stringType()
28+
.noDefaultValue()
29+
.withDescription(
30+
"your infura project url like : https://mainnet.infura.io/v3/xxxxxxxxxxxx");
31+
}

‎seatunnel-connectors-v2/connector-web3j/src/main/java/org/apache/seatunnel/connectors/seatunnel/source/Web3jSource.java

+85
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,85 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.seatunnel.connectors.seatunnel.source;
19+
20+
import org.apache.seatunnel.api.common.JobContext;
21+
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
22+
import org.apache.seatunnel.api.source.Boundedness;
23+
import org.apache.seatunnel.api.table.catalog.CatalogTable;
24+
import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
25+
import org.apache.seatunnel.api.table.catalog.TableIdentifier;
26+
import org.apache.seatunnel.api.table.catalog.TablePath;
27+
import org.apache.seatunnel.api.table.catalog.TableSchema;
28+
import org.apache.seatunnel.api.table.type.BasicType;
29+
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
30+
import org.apache.seatunnel.common.constants.JobMode;
31+
import org.apache.seatunnel.connectors.seatunnel.common.source.AbstractSingleSplitReader;
32+
import org.apache.seatunnel.connectors.seatunnel.common.source.AbstractSingleSplitSource;
33+
import org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplitReaderContext;
34+
35+
import java.util.ArrayList;
36+
import java.util.Collections;
37+
import java.util.HashMap;
38+
import java.util.List;
39+
40+
public class Web3jSource extends AbstractSingleSplitSource<SeaTunnelRow> {
41+
private Web3jSourceParameter parameter;
42+
private JobContext jobContext;
43+
44+
public Web3jSource(ReadonlyConfig readonlyConfig) {
45+
this.parameter = new Web3jSourceParameter(readonlyConfig);
46+
}
47+
48+
@Override
49+
public Boundedness getBoundedness() {
50+
return JobMode.BATCH.equals(jobContext.getJobMode())
51+
? Boundedness.BOUNDED
52+
: Boundedness.UNBOUNDED;
53+
}
54+
55+
@Override
56+
public String getPluginName() {
57+
return "Web3j";
58+
}
59+
60+
@Override
61+
public void setJobContext(JobContext jobContext) {
62+
this.jobContext = jobContext;
63+
}
64+
65+
@Override
66+
public List<CatalogTable> getProducedCatalogTables() {
67+
return Collections.singletonList(
68+
CatalogTable.of(
69+
TableIdentifier.of("Web3j", TablePath.DEFAULT),
70+
TableSchema.builder()
71+
.column(
72+
PhysicalColumn.of(
73+
"value", BasicType.STRING_TYPE, 0L, true, null, ""))
74+
.build(),
75+
new HashMap<>(),
76+
new ArrayList<>(),
77+
""));
78+
}
79+
80+
@Override
81+
public AbstractSingleSplitReader<SeaTunnelRow> createReader(
82+
SingleSplitReaderContext readerContext) throws Exception {
83+
return new Web3jSourceReader(this.parameter, readerContext);
84+
}
85+
}

‎seatunnel-connectors-v2/connector-web3j/src/main/java/org/apache/seatunnel/connectors/seatunnel/source/Web3jSourceFactory.java

+56
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,56 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.seatunnel.connectors.seatunnel.source;
19+
20+
import org.apache.seatunnel.api.configuration.util.OptionRule;
21+
import org.apache.seatunnel.api.source.SeaTunnelSource;
22+
import org.apache.seatunnel.api.source.SourceSplit;
23+
import org.apache.seatunnel.api.table.connector.TableSource;
24+
import org.apache.seatunnel.api.table.factory.Factory;
25+
import org.apache.seatunnel.api.table.factory.TableSourceFactory;
26+
import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
27+
28+
import com.google.auto.service.AutoService;
29+
30+
import java.io.Serializable;
31+
32+
import static org.apache.seatunnel.connectors.seatunnel.config.Web3jConfig.URL;
33+
34+
@AutoService(Factory.class)
35+
public class Web3jSourceFactory implements TableSourceFactory {
36+
@Override
37+
public String factoryIdentifier() {
38+
return "Web3j";
39+
}
40+
41+
@Override
42+
public OptionRule optionRule() {
43+
return OptionRule.builder().required(URL).build();
44+
}
45+
46+
@Override
47+
public Class<? extends SeaTunnelSource> getSourceClass() {
48+
return Web3jSource.class;
49+
}
50+
51+
@Override
52+
public <T, SplitT extends SourceSplit, StateT extends Serializable>
53+
TableSource<T, SplitT, StateT> createSource(TableSourceFactoryContext context) {
54+
return () -> (SeaTunnelSource<T, SplitT, StateT>) new Web3jSource(context.getOptions());
55+
}
56+
}

‎seatunnel-connectors-v2/connector-web3j/src/main/java/org/apache/seatunnel/connectors/seatunnel/source/Web3jSourceParameter.java

+36
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.seatunnel.connectors.seatunnel.source;
19+
20+
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
21+
22+
import java.io.Serializable;
23+
24+
import static org.apache.seatunnel.connectors.seatunnel.config.Web3jConfig.URL;
25+
26+
public class Web3jSourceParameter implements Serializable {
27+
private final String url;
28+
29+
public String getUrl() {
30+
return url;
31+
}
32+
33+
public Web3jSourceParameter(ReadonlyConfig config) {
34+
this.url = config.get(URL);
35+
}
36+
}

0 commit comments

Comments
 (0)