Skip to content

Commit 85aae47

Browse files
author
xuchao
committed
回滚 5f30a74 ，解决 '回滚导致的test分支变更内容'无法提交的问题
1 parent b59ef71 commit 85aae47

File tree

256 files changed

+3487
-2621
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

256 files changed

+3487
-2621
lines changed

‎.gitlab-ci.yml‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
build:
2+
stage: test
3+
script:
4+
- mvn clean org.jacoco:jacoco-maven-plugin:0.7.8:prepare-agent package -Dmaven.test.failure.ignore=true -q
5+
- mvn sonar:sonar -Dsonar.projectKey="dt-insight-engine/flinkStreamSQL" -Dsonar.branch.name="v1.8.0_dev" -Dsonar.login=11974c5e9a29625efa09fdc3c3fdc031efb1aab1 -Dsonar.host.url=http://172.16.100.198:9000 -Dsonar.jdbc.url=jdbc:postgresql://172.16.100.198:5432/sonar -Dsonar.java.binaries=target/sonar
6+
- sh ci/sonar_notify.sh
7+
only:
8+
- v1.8.0_dev
9+
tags:
10+
- dt-insight-engine

‎cassandra/cassandra-side/cassandra-all-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAllReqRow.java‎

Lines changed: 4 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -34,10 +34,10 @@
3434
import com.datastax.driver.core.SocketOptions;
3535
import com.datastax.driver.core.policies.DowngradingConsistencyRetryPolicy;
3636
import com.datastax.driver.core.policies.RetryPolicy;
37-
import com.dtstack.flink.sql.side.AllReqRow;
37+
import com.dtstack.flink.sql.side.BaseAllReqRow;
3838
import com.dtstack.flink.sql.side.FieldInfo;
3939
import com.dtstack.flink.sql.side.JoinInfo;
40-
import com.dtstack.flink.sql.side.SideTableInfo;
40+
import com.dtstack.flink.sql.side.AbstractSideTableInfo;
4141
import com.dtstack.flink.sql.side.cassandra.table.CassandraSideTableInfo;
4242
import com.google.common.collect.Lists;
4343
import com.google.common.collect.Maps;
@@ -62,14 +62,12 @@
6262
*
6363
* @author xuqianjin
6464
*/
65-
public class CassandraAllReqRow extends AllReqRow {
65+
public class CassandraAllReqRow extends BaseAllReqRow {
6666

6767
private static final long serialVersionUID = 54015343561288219L;
6868

6969
private static final Logger LOG = LoggerFactory.getLogger(CassandraAllReqRow.class);
7070

71-
private static final String cassandra_DRIVER = "com.cassandra.jdbc.Driver";
72-
7371
private static final int CONN_RETRY_NUM = 3;
7472

7573
private static final int FETCH_SIZE = 1000;
@@ -79,7 +77,7 @@ public class CassandraAllReqRow extends AllReqRow {
7977

8078
private AtomicReference<Map<String, List<Map<String, Object>>>> cacheRef = new AtomicReference<>();
8179

82-
public CassandraAllReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
80+
public CassandraAllReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, AbstractSideTableInfo sideTableInfo) {
8381
super(new com.dtstack.flink.sql.side.cassandra.CassandraAllSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));
8482
}
8583

‎cassandra/cassandra-side/cassandra-all-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAllSideInfo.java‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -20,8 +20,8 @@
2020

2121
import com.dtstack.flink.sql.side.FieldInfo;
2222
import com.dtstack.flink.sql.side.JoinInfo;
23-
import com.dtstack.flink.sql.side.SideInfo;
24-
import com.dtstack.flink.sql.side.SideTableInfo;
23+
import com.dtstack.flink.sql.side.BaseSideInfo;
24+
import com.dtstack.flink.sql.side.AbstractSideTableInfo;
2525
import com.dtstack.flink.sql.side.cassandra.table.CassandraSideTableInfo;
2626
import com.dtstack.flink.sql.util.ParseUtils;
2727
import org.apache.calcite.sql.SqlNode;
@@ -37,16 +37,16 @@
3737
*
3838
* @author xuqianjin
3939
*/
40-
public class CassandraAllSideInfo extends SideInfo {
40+
public class CassandraAllSideInfo extends BaseSideInfo {
4141

4242
private static final long serialVersionUID = -8690814317653033557L;
4343

44-
public CassandraAllSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
44+
public CassandraAllSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, AbstractSideTableInfo sideTableInfo) {
4545
super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
4646
}
4747

4848
@Override
49-
public void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo) {
49+
public void buildEqualInfo(JoinInfo joinInfo, AbstractSideTableInfo sideTableInfo) {
5050
CassandraSideTableInfo cassandraSideTableInfo = (CassandraSideTableInfo) sideTableInfo;
5151

5252
sqlCondition = "select ${selectField} from ${tableName} ";

‎cassandra/cassandra-side/cassandra-async-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAsyncReqRow.java‎

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -37,11 +37,11 @@
3737
import com.datastax.driver.core.policies.DowngradingConsistencyRetryPolicy;
3838
import com.datastax.driver.core.policies.RetryPolicy;
3939
import com.dtstack.flink.sql.enums.ECacheContentType;
40-
import com.dtstack.flink.sql.side.AsyncReqRow;
40+
import com.dtstack.flink.sql.side.BaseAsyncReqRow;
4141
import com.dtstack.flink.sql.side.CacheMissVal;
4242
import com.dtstack.flink.sql.side.FieldInfo;
4343
import com.dtstack.flink.sql.side.JoinInfo;
44-
import com.dtstack.flink.sql.side.SideTableInfo;
44+
import com.dtstack.flink.sql.side.AbstractSideTableInfo;
4545
import com.dtstack.flink.sql.side.cache.CacheObj;
4646
import com.dtstack.flink.sql.side.cassandra.table.CassandraSideTableInfo;
4747
import com.google.common.base.Function;
@@ -67,7 +67,7 @@
6767
*
6868
* @author xuqianjin
6969
*/
70-
public class CassandraAsyncReqRow extends AsyncReqRow {
70+
public class CassandraAsyncReqRow extends BaseAsyncReqRow {
7171

7272
private static final long serialVersionUID = 6631584128079864735L;
7373

@@ -83,7 +83,7 @@ public class CassandraAsyncReqRow extends AsyncReqRow {
8383
private transient ListenableFuture session;
8484
private transient CassandraSideTableInfo cassandraSideTableInfo;
8585

86-
public CassandraAsyncReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
86+
public CassandraAsyncReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, AbstractSideTableInfo sideTableInfo) {
8787
super(new com.dtstack.flink.sql.side.cassandra.CassandraAsyncSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));
8888
}
8989

@@ -216,7 +216,7 @@ public void asyncInvoke(CRow input, ResultFuture<CRow> resultFuture) throws Exce
216216
connCassandraDB(cassandraSideTableInfo);
217217

218218
String sqlCondition = sideInfo.getSqlCondition() + " " + sqlWhere + " ALLOW FILTERING ";
219-
System.out.println("sqlCondition:" + sqlCondition);
219+
LOG.info("sqlCondition:{}" + sqlCondition);
220220

221221
ListenableFuture<ResultSet> resultSet = Futures.transformAsync(session,
222222
new AsyncFunction<Session, ResultSet>() {
@@ -265,7 +265,6 @@ public void onSuccess(List<com.datastax.driver.core.Row> rows) {
265265
public void onFailure(Throwable t) {
266266
LOG.error("Failed to retrieve the data: %s%n",
267267
t.getMessage());
268-
System.out.println("Failed to retrieve the data: " + t.getMessage());
269268
cluster.closeAsync();
270269
resultFuture.completeExceptionally(t);
271270
}

‎cassandra/cassandra-side/cassandra-async-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAsyncSideInfo.java‎

Lines changed: 11 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -20,8 +20,8 @@
2020

2121
import com.dtstack.flink.sql.side.FieldInfo;
2222
import com.dtstack.flink.sql.side.JoinInfo;
23-
import com.dtstack.flink.sql.side.SideInfo;
24-
import com.dtstack.flink.sql.side.SideTableInfo;
23+
import com.dtstack.flink.sql.side.BaseSideInfo;
24+
import com.dtstack.flink.sql.side.AbstractSideTableInfo;
2525
import com.dtstack.flink.sql.side.cassandra.table.CassandraSideTableInfo;
2626
import com.dtstack.flink.sql.util.ParseUtils;
2727
import org.apache.calcite.sql.SqlBasicCall;
@@ -30,6 +30,8 @@
3030
import org.apache.calcite.sql.SqlNode;
3131
import org.apache.flink.api.java.typeutils.RowTypeInfo;
3232
import com.google.common.collect.Lists;
33+
import org.slf4j.Logger;
34+
import org.slf4j.LoggerFactory;
3335

3436
import java.util.List;
3537

@@ -39,16 +41,18 @@
3941
*
4042
* @author xuqianjin
4143
*/
42-
public class CassandraAsyncSideInfo extends SideInfo {
44+
public class CassandraAsyncSideInfo extends BaseSideInfo {
4345

4446
private static final long serialVersionUID = -4403313049809013362L;
47+
private static final Logger LOG = LoggerFactory.getLogger(CassandraAsyncSideInfo.class.getSimpleName());
4548

46-
public CassandraAsyncSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
49+
50+
public CassandraAsyncSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, AbstractSideTableInfo sideTableInfo) {
4751
super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
4852
}
4953

5054
@Override
51-
public void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo) {
55+
public void buildEqualInfo(JoinInfo joinInfo, AbstractSideTableInfo sideTableInfo) {
5256
CassandraSideTableInfo cassandraSideTableInfo = (CassandraSideTableInfo) sideTableInfo;
5357

5458
String sideTableName = joinInfo.getSideTableName();
@@ -63,9 +67,9 @@ public void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo) {
6367
}
6468

6569
sqlCondition = "select ${selectField} from ${tableName}";
66-
6770
sqlCondition = sqlCondition.replace("${tableName}", cassandraSideTableInfo.getDatabase()+"."+cassandraSideTableInfo.getTableName()).replace("${selectField}", sideSelectFields);
68-
System.out.println("---------side_exe_sql-----\n" + sqlCondition);
71+
72+
LOG.info("---------side_exe_sql-----\n{}" + sqlCondition);
6973
}
7074

7175

‎cassandra/cassandra-side/cassandra-side-core/src/main/java/com/dtstack/flink/sql/side/cassandra/table/CassandraSideParser.java‎

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -19,26 +19,24 @@
1919

2020
package com.dtstack.flink.sql.side.cassandra.table;
2121

22-
import com.dtstack.flink.sql.table.AbsSideTableParser;
23-
import com.dtstack.flink.sql.table.TableInfo;
22+
import com.dtstack.flink.sql.table.AbstractSideTableParser;
23+
import com.dtstack.flink.sql.table.AbstractTableInfo;
2424
import com.dtstack.flink.sql.util.MathUtil;
2525

26-
import java.math.BigDecimal;
27-
import java.sql.Date;
2826
import java.sql.Timestamp;
2927
import java.util.Map;
3028
import java.util.regex.Matcher;
3129
import java.util.regex.Pattern;
3230

33-
import static com.dtstack.flink.sql.table.TableInfo.PARALLELISM_KEY;
31+
import static com.dtstack.flink.sql.table.AbstractTableInfo.PARALLELISM_KEY;
3432

3533
/**
3634
* Reason:
3735
* Date: 2018/11/22
3836
*
3937
* @author xuqianjin
4038
*/
41-
public class CassandraSideParser extends AbsSideTableParser {
39+
public class CassandraSideParser extends AbstractSideTableParser {
4240

4341
private final static String SIDE_SIGN_KEY = "sideSignKey";
4442

@@ -73,7 +71,7 @@ public CassandraSideParser() {
7371
}
7472

7573
@Override
76-
public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, Object> props) {
74+
public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, Object> props) {
7775
com.dtstack.flink.sql.side.cassandra.table.CassandraSideTableInfo cassandraSideTableInfo = new com.dtstack.flink.sql.side.cassandra.table.CassandraSideTableInfo();
7876
cassandraSideTableInfo.setName(tableName);
7977
parseFieldsInfo(fieldsInfo, cassandraSideTableInfo);
@@ -96,9 +94,10 @@ public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, O
9694
return cassandraSideTableInfo;
9795
}
9896

99-
private void dealSideSign(Matcher matcher, TableInfo tableInfo) {
97+
private void dealSideSign(Matcher matcher, AbstractTableInfo tableInfo) {
10098
}
10199

100+
@Override
102101
public Class dbTypeConvertToJavaType(String fieldType) {
103102
switch (fieldType.toLowerCase()) {
104103
case "bigint":
@@ -121,6 +120,8 @@ public Class dbTypeConvertToJavaType(String fieldType) {
121120
return Double.class;
122121
case "timestamp":
123122
return Timestamp.class;
123+
default:
124+
break;
124125
}
125126

126127
throw new RuntimeException("不支持 " + fieldType + " 类型");

‎cassandra/cassandra-side/cassandra-side-core/src/main/java/com/dtstack/flink/sql/side/cassandra/table/CassandraSideTableInfo.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@
1919

2020
package com.dtstack.flink.sql.side.cassandra.table;
2121

22-
import com.dtstack.flink.sql.side.SideTableInfo;
22+
import com.dtstack.flink.sql.side.AbstractSideTableInfo;
2323
import com.google.common.base.Preconditions;
2424

2525
/**
@@ -28,7 +28,7 @@
2828
*
2929
* @author xuqianjin
3030
*/
31-
public class CassandraSideTableInfo extends SideTableInfo {
31+
public class CassandraSideTableInfo extends AbstractSideTableInfo {
3232

3333
private static final long serialVersionUID = -5556431094535478915L;
3434

‎cassandra/cassandra-sink/src/main/java/com/dtstack/flink/sql/sink/cassandra/CassandraOutputFormat.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -54,7 +54,7 @@
5454
import com.datastax.driver.core.SocketOptions;
5555
import com.datastax.driver.core.policies.DowngradingConsistencyRetryPolicy;
5656
import com.datastax.driver.core.policies.RetryPolicy;
57-
import com.dtstack.flink.sql.outputformat.DtRichOutputFormat;
57+
import com.dtstack.flink.sql.outputformat.AbstractDtRichOutputFormat;
5858
import org.apache.commons.lang3.StringUtils;
5959
import org.slf4j.Logger;
6060
import org.slf4j.LoggerFactory;
@@ -72,7 +72,7 @@
7272
* @see Tuple
7373
* @see DriverManager
7474
*/
75-
public class CassandraOutputFormat extends DtRichOutputFormat<Tuple2> {
75+
public class CassandraOutputFormat extends AbstractDtRichOutputFormat<Tuple2> {
7676
private static final long serialVersionUID = -7994311331389155692L;
7777

7878
private static final Logger LOG = LoggerFactory.getLogger(CassandraOutputFormat.class);

‎cassandra/cassandra-sink/src/main/java/com/dtstack/flink/sql/sink/cassandra/CassandraSink.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@
2222

2323
import com.dtstack.flink.sql.sink.IStreamSinkGener;
2424
import com.dtstack.flink.sql.sink.cassandra.table.CassandraTableInfo;
25-
import com.dtstack.flink.sql.table.TargetTableInfo;
25+
import com.dtstack.flink.sql.table.AbstractTargetTableInfo;
2626
import org.apache.flink.api.common.typeinfo.TypeInformation;
2727
import org.apache.flink.api.java.tuple.Tuple2;
2828
import org.apache.flink.api.java.typeutils.RowTypeInfo;
@@ -63,7 +63,7 @@ public CassandraSink() {
6363
}
6464

6565
@Override
66-
public CassandraSink genStreamSink(TargetTableInfo targetTableInfo) {
66+
public CassandraSink genStreamSink(AbstractTargetTableInfo targetTableInfo) {
6767
CassandraTableInfo cassandraTableInfo = (CassandraTableInfo) targetTableInfo;
6868
this.address = cassandraTableInfo.getAddress();
6969
this.tableName = cassandraTableInfo.getTableName();

‎cassandra/cassandra-sink/src/main/java/com/dtstack/flink/sql/sink/cassandra/table/CassandraSinkParser.java‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -19,21 +19,21 @@
1919

2020
package com.dtstack.flink.sql.sink.cassandra.table;
2121

22-
import com.dtstack.flink.sql.table.AbsTableParser;
23-
import com.dtstack.flink.sql.table.TableInfo;
22+
import com.dtstack.flink.sql.table.AbstractTableParser;
23+
import com.dtstack.flink.sql.table.AbstractTableInfo;
2424
import com.dtstack.flink.sql.util.MathUtil;
2525

2626
import java.util.Map;
2727

28-
import static com.dtstack.flink.sql.table.TableInfo.PARALLELISM_KEY;
28+
import static com.dtstack.flink.sql.table.AbstractTableInfo.PARALLELISM_KEY;
2929

3030
/**
3131
* Reason:
3232
* Date: 2018/11/22
3333
*
3434
* @author xuqianjin
3535
*/
36-
public class CassandraSinkParser extends AbsTableParser {
36+
public class CassandraSinkParser extends AbstractTableParser {
3737

3838
public static final String ADDRESS_KEY = "address";
3939

@@ -60,7 +60,7 @@ public class CassandraSinkParser extends AbsTableParser {
6060
public static final String POOL_TIMEOUT_MILLIS_KEY = "poolTimeoutMillis";
6161

6262
@Override
63-
public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, Object> props) {
63+
public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, Object> props) {
6464
CassandraTableInfo cassandraTableInfo = new CassandraTableInfo();
6565
cassandraTableInfo.setName(tableName);
6666
parseFieldsInfo(fieldsInfo, cassandraTableInfo);

0 commit comments

Comments
 (0)