Skip to content

Commit a7804c3

Browse files
loserwang1024loserwang1024
committed
add log to find problem
1 parent 1be6d2d commit a7804c3

File tree

8 files changed

+28
-0
lines changed

8 files changed

+28
-0
lines changed

‎flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/com/ververica/cdc/connectors/base/source/reader/external/IncrementalSourceScanFetcher.java

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -150,6 +150,7 @@ public Iterator<SourceRecords> pollSplitRecords() throws InterruptedException {
150150
if (!reachChangeLogStart) {
151151
outputBuffer.put((Struct) record.key(), record);
152152
} else {
153+
LOG.info("get backfill data here: " + record.toString());
153154
if (isChangeRecordInChunkRange(record)) {
154155
// rewrite overlapping snapshot records through the record key
155156
taskContext.rewriteOutputBuffer(outputBuffer, record);
@@ -242,6 +243,8 @@ private boolean isChangeRecordInChunkRange(SourceRecord record) {
242243
currentSnapshotSplit.getSplitStart(),
243244
currentSnapshotSplit.getSplitEnd());
244245
}
246+
247+
LOG.info("isChangeRecordInChunkRange filter record out here: " + record);
245248
return false;
246249
}
247250
}

‎flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/main/java/com/ververica/cdc/connectors/oracle/source/reader/fetch/OracleScanFetchTask.java

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,7 @@
5858

5959
/** The task to work for fetching data of Oracle table snapshot split. */
6060
public class OracleScanFetchTask extends AbstractScanFetchTask {
61+
private static final Logger LOG = LoggerFactory.getLogger(OracleScanFetchTask.class);
6162

6263
public OracleScanFetchTask(SnapshotSplit split) {
6364
super(split);
@@ -92,6 +93,7 @@ protected void executeDataSnapshot(Context context) throws Exception {
9293
@Override
9394
protected void executeBackfillTask(Context context, StreamSplit backfillStreamSplit)
9495
throws Exception {
96+
LOG.info("executeBackfillTask here");
9597
OracleSourceFetchTaskContext sourceFetchContext = (OracleSourceFetchTaskContext) context;
9698

9799
final RedoLogSplitReadTask backfillRedoLogReadTask =

‎flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/main/java/com/ververica/cdc/connectors/oracle/source/reader/fetch/OracleStreamFetchTask.java

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -141,6 +141,10 @@ protected void afterHandleScn(
141141
if (isBoundedRead()) {
142142
final RedoLogOffset currentRedoLogOffset =
143143
getCurrentRedoLogOffset(offsetContext.getOffset());
144+
LOG.info(
145+
"afterHandleScn read streaming data here: "
146+
+ currentRedoLogOffset.toString());
147+
144148
// reach the high watermark, the redo log fetcher should be finished
145149
if (currentRedoLogOffset.isAtOrAfter(redoLogSplit.getEndingOffset())) {
146150
// send redo log end event

‎flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/test/java/com/ververica/cdc/connectors/oracle/OracleChangeEventSourceExampleTest.java

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@
3737
import java.util.Properties;
3838

3939
/** Example Tests for {@link JdbcIncrementalSource}. */
40+
@Ignore
4041
public class OracleChangeEventSourceExampleTest extends OracleSourceTestBase {
4142

4243
private static final Logger LOG =

‎flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/test/java/com/ververica/cdc/connectors/oracle/OracleSourceTest.java

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,7 @@
6868
import static org.junit.Assert.assertTrue;
6969

7070
/** Tests for {@link OracleSource} which also heavily tests {@link DebeziumSourceFunction}. */
71+
@Ignore
7172
public class OracleSourceTest extends OracleSourceTestBase {
7273

7374
private static final Logger LOG = LoggerFactory.getLogger(OracleSourceTest.class);

‎flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/test/java/com/ververica/cdc/connectors/oracle/source/OracleSourceITCase.java

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@
3939
import io.debezium.connector.oracle.OracleConnection;
4040
import io.debezium.jdbc.JdbcConfiguration;
4141
import org.apache.commons.lang3.StringUtils;
42+
import org.junit.Ignore;
4243
import org.junit.Rule;
4344
import org.junit.Test;
4445
import org.junit.rules.Timeout;
@@ -71,55 +72,64 @@ public class OracleSourceITCase extends OracleSourceTestBase {
7172
@Rule public final Timeout timeoutPerTest = Timeout.seconds(300);
7273

7374
@Test
75+
@Ignore
7476
public void testReadSingleTableWithSingleParallelism() throws Exception {
7577
testOracleParallelSource(
7678
1, FailoverType.NONE, FailoverPhase.NEVER, new String[] {"CUSTOMERS"});
7779
}
7880

7981
@Test
82+
@Ignore
8083
public void testReadSingleTableWithMultipleParallelism() throws Exception {
8184
testOracleParallelSource(
8285
4, FailoverType.NONE, FailoverPhase.NEVER, new String[] {"CUSTOMERS"});
8386
}
8487

8588
// Failover tests
8689
@Test
90+
@Ignore
8791
public void testTaskManagerFailoverInSnapshotPhase() throws Exception {
8892
testOracleParallelSource(
8993
FailoverType.TM, FailoverPhase.SNAPSHOT, new String[] {"CUSTOMERS"});
9094
}
9195

9296
@Test
97+
@Ignore
9398
public void testTaskManagerFailoverInRedoLogPhase() throws Exception {
9499
testOracleParallelSource(
95100
FailoverType.TM, FailoverPhase.REDO_LOG, new String[] {"CUSTOMERS"});
96101
}
97102

98103
@Test
104+
@Ignore
99105
public void testJobManagerFailoverInSnapshotPhase() throws Exception {
100106
testOracleParallelSource(
101107
FailoverType.JM, FailoverPhase.SNAPSHOT, new String[] {"CUSTOMERS"});
102108
}
103109

104110
@Test
111+
@Ignore
105112
public void testJobManagerFailoverInRedoLogPhase() throws Exception {
106113
testOracleParallelSource(
107114
FailoverType.JM, FailoverPhase.REDO_LOG, new String[] {"CUSTOMERS"});
108115
}
109116

110117
@Test
118+
@Ignore
111119
public void testTaskManagerFailoverSingleParallelism() throws Exception {
112120
testOracleParallelSource(
113121
1, FailoverType.TM, FailoverPhase.SNAPSHOT, new String[] {"CUSTOMERS"});
114122
}
115123

116124
@Test
125+
@Ignore
117126
public void testJobManagerFailoverSingleParallelism() throws Exception {
118127
testOracleParallelSource(
119128
1, FailoverType.JM, FailoverPhase.SNAPSHOT, new String[] {"CUSTOMERS"});
120129
}
121130

122131
@Test
132+
@Ignore
123133
public void testReadSingleTableWithSingleParallelismAndSkipBackfill() throws Exception {
124134
testOracleParallelSource(
125135
DEFAULT_PARALLELISM,
@@ -166,6 +176,7 @@ public void testEnableBackfillWithPreHighWaterMark() throws Exception {
166176
}
167177

168178
@Test
179+
@Ignore
169180
public void testEnableBackfillWithPostLowWaterMark() throws Exception {
170181
List<String> records = getResultOfWithHooks(false, 21, USE_POST_LOWWATERMARK_HOOK);
171182

@@ -198,6 +209,7 @@ public void testEnableBackfillWithPostLowWaterMark() throws Exception {
198209
}
199210

200211
@Test
212+
@Ignore
201213
public void testSkipBackfillWithPreHighWaterMark() throws Exception {
202214
List<String> records = getResultOfWithHooks(true, 25, USE_PRE_HIGHWATERMARK_HOOK);
203215

@@ -234,6 +246,7 @@ public void testSkipBackfillWithPreHighWaterMark() throws Exception {
234246
}
235247

236248
@Test
249+
@Ignore
237250
public void testSkipBackfillWithPostLowWaterMark() throws Exception {
238251

239252
List<String> records = getResultOfWithHooks(true, 25, USE_POST_LOWWATERMARK_HOOK);

‎flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/test/java/com/ververica/cdc/connectors/oracle/table/OracleConnectorITCase.java

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
import org.junit.After;
2929
import org.junit.Assume;
3030
import org.junit.Before;
31+
import org.junit.Ignore;
3132
import org.junit.Test;
3233
import org.junit.runner.RunWith;
3334
import org.junit.runners.Parameterized;
@@ -65,6 +66,7 @@
6566

6667
/** Integration tests for Oracle redo log SQL source. */
6768
@RunWith(Parameterized.class)
69+
@Ignore
6870
public class OracleConnectorITCase {
6971
private static final int RECORDS_COUNT = 10_000;
7072
private static final int WORKERS_COUNT = 4;

‎flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/test/java/com/ververica/cdc/connectors/oracle/table/OracleTableSourceFactoryTest.java

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@
3434
import com.ververica.cdc.connectors.base.options.JdbcSourceOptions;
3535
import com.ververica.cdc.connectors.base.options.SourceOptions;
3636
import com.ververica.cdc.connectors.base.options.StartupOptions;
37+
import org.junit.Ignore;
3738
import org.junit.Test;
3839

3940
import java.time.Duration;
@@ -52,6 +53,7 @@
5253
* Test for {@link com.ververica.cdc.connectors.oracle.table.OracleTableSource} created by {@link
5354
* com.ververica.cdc.connectors.oracle.table.OracleTableSourceFactory}.
5455
*/
56+
@Ignore
5557
public class OracleTableSourceFactoryTest {
5658
private static final ResolvedSchema SCHEMA =
5759
new ResolvedSchema(

0 commit comments

Comments
 (0)