Skip to content

Commit 4e0663a

Browse files
author
Aditya Bharadwaj
committed
corrected connector name
1 parent f7e7462 commit 4e0663a

12 files changed

Lines changed: 81 additions & 76 deletions

File tree

‎v2/datastream-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/DataStreamToSpanner.java‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -45,7 +45,7 @@
4545
import com.google.cloud.teleport.v2.spanner.migrations.utils.TransformationContextReader;
4646
import com.google.cloud.teleport.v2.templates.DataStreamToSpanner.Options;
4747
import com.google.cloud.teleport.v2.templates.constants.DatastreamToSpannerConstants;
48-
import com.google.cloud.teleport.v2.templates.datastream.source.SourceConnectorRegistry;
48+
import com.google.cloud.teleport.v2.templates.datastream.source.DatastreamToSpannerSourceConnectorRegistry;
4949
import com.google.cloud.teleport.v2.templates.spanner.ProcessInformationSchema;
5050
import com.google.cloud.teleport.v2.templates.transform.ChangeEventTransformerDoFn;
5151
import com.google.cloud.teleport.v2.transforms.DLQWriteTransform;
@@ -589,12 +589,13 @@ static void validateSourceType(Options options) {
589589
return;
590590
}
591591
String sourceType = getSourceType(options);
592-
if (!SourceConnectorRegistry.getSupportedSourceTypes().contains(sourceType)) {
592+
if (!DatastreamToSpannerSourceConnectorRegistry.getSupportedSourceTypes()
593+
.contains(sourceType)) {
593594
throw new IllegalArgumentException(
594595
"Unsupported source type found: "
595596
+ sourceType
596597
+ ". Specify one of the following source types: "
597-
+ SourceConnectorRegistry.getSupportedSourceTypes());
598+
+ DatastreamToSpannerSourceConnectorRegistry.getSupportedSourceTypes());
598599
}
599600
options.setDatastreamSourceType(sourceType);
600601
}
@@ -621,7 +622,7 @@ static String getSourceType(Options options) {
621622

622623
static String getSourceTypeFromConfig(SourceConfig sourceConfig) {
623624
try {
624-
return SourceConnectorRegistry.getSourceTypeFromConfig(sourceConfig);
625+
return DatastreamToSpannerSourceConnectorRegistry.getSourceTypeFromConfig(sourceConfig);
625626
} catch (IllegalArgumentException e) {
626627
LOG.error("Source Connection Profile Type Not Supported", e);
627628
throw e;

‎v2/datastream-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/SpannerTransactionWriterDoFn.java‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -40,8 +40,8 @@
4040
import com.google.cloud.teleport.v2.templates.datastream.ChangeEventContext;
4141
import com.google.cloud.teleport.v2.templates.datastream.ChangeEventSequence;
4242
import com.google.cloud.teleport.v2.templates.datastream.DatastreamConstants;
43-
import com.google.cloud.teleport.v2.templates.datastream.source.ISourceConnector;
44-
import com.google.cloud.teleport.v2.templates.datastream.source.SourceConnectorRegistry;
43+
import com.google.cloud.teleport.v2.templates.datastream.source.DatastreamToSpannerSourceConnectorRegistry;
44+
import com.google.cloud.teleport.v2.templates.datastream.source.IDsToSpSourceConnector;
4545
import com.google.cloud.teleport.v2.templates.spanner.DatastreamToSpannerExceptionClassifier;
4646
import com.google.cloud.teleport.v2.templates.spanner.DatastreamToSpannerExceptionClassifier.ErrorTag;
4747
import com.google.cloud.teleport.v2.templates.utils.WatchdogRunnable;
@@ -105,7 +105,7 @@ class SpannerTransactionWriterDoFn
105105
/* SpannerAccessor must be transient so that its value is not serialized at runtime. */
106106
private transient SpannerAccessor spannerAccessor;
107107

108-
private transient ISourceConnector sourceConnector;
108+
private transient IDsToSpSourceConnector sourceConnector;
109109
/* SpannerAccessor for shadow table database must be transient so that its value is not serialized at runtime. */
110110
private transient SpannerAccessor shadowTableSpannerAccessor;
111111

@@ -219,7 +219,7 @@ public void setup() {
219219
: spannerAccessor;
220220
mapper = new ObjectMapper();
221221
mapper.enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS);
222-
sourceConnector = SourceConnectorRegistry.getSourceConnector(sourceType);
222+
sourceConnector = DatastreamToSpannerSourceConnectorRegistry.getSourceConnector(sourceType);
223223
// Setup and start the watchdog thread.
224224
transactionAttemptCount = new AtomicLong(0);
225225
isInTransaction = new AtomicBoolean(false);
@@ -598,14 +598,14 @@ String getTxnTag(PipelineOptions options) {
598598
return txnTag;
599599
}
600600

601-
private ISourceConnector getSourceConnector() {
601+
private IDsToSpSourceConnector getSourceConnector() {
602602
if (sourceConnector == null) {
603-
sourceConnector = SourceConnectorRegistry.getSourceConnector(sourceType);
603+
sourceConnector = DatastreamToSpannerSourceConnectorRegistry.getSourceConnector(sourceType);
604604
}
605605
return sourceConnector;
606606
}
607607

608-
void setSourceConnector(ISourceConnector sourceConnector) {
608+
void setSourceConnector(IDsToSpSourceConnector sourceConnector) {
609609
this.sourceConnector = sourceConnector;
610610
}
611611

v2/datastream-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/datastream/source/SourceConnectorRegistry.java renamed to v2/datastream-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/datastream/source/DatastreamToSpannerSourceConnectorRegistry.java

Lines changed: 17 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -16,9 +16,9 @@
1616
package com.google.cloud.teleport.v2.templates.datastream.source;
1717

1818
import com.google.api.services.datastream.v1.model.SourceConfig;
19-
import com.google.cloud.teleport.v2.templates.datastream.source.mysql.MySqlSourceConnector;
20-
import com.google.cloud.teleport.v2.templates.datastream.source.oracle.OracleSourceConnector;
21-
import com.google.cloud.teleport.v2.templates.datastream.source.postgresql.PostgresqlSourceConnector;
19+
import com.google.cloud.teleport.v2.templates.datastream.source.mysql.MySqlDsToSpSourceConnector;
20+
import com.google.cloud.teleport.v2.templates.datastream.source.oracle.OracleDsToSpSourceConnector;
21+
import com.google.cloud.teleport.v2.templates.datastream.source.postgresql.PostgresqlDsToSpSourceConnector;
2222
import java.util.Collection;
2323
import java.util.Collections;
2424
import java.util.HashMap;
@@ -27,21 +27,20 @@
2727
import org.slf4j.Logger;
2828
import org.slf4j.LoggerFactory;
2929

30-
/**
31-
* Registry that automatically discovers and loads {@link ISourceConnector} connectors using SPI.
32-
*/
33-
public class SourceConnectorRegistry {
30+
/** Registry for datastream to spanner source connectors */
31+
public class DatastreamToSpannerSourceConnectorRegistry {
3432

35-
private static final Logger LOG = LoggerFactory.getLogger(SourceConnectorRegistry.class);
36-
private static final Map<String, ISourceConnector> CONNECTORS = new HashMap<>();
33+
private static final Logger LOG =
34+
LoggerFactory.getLogger(DatastreamToSpannerSourceConnectorRegistry.class);
35+
private static final Map<String, IDsToSpSourceConnector> CONNECTORS = new HashMap<>();
3736

3837
static {
39-
register(new MySqlSourceConnector());
40-
register(new PostgresqlSourceConnector());
41-
register(new OracleSourceConnector());
38+
register(new MySqlDsToSpSourceConnector());
39+
register(new PostgresqlDsToSpSourceConnector());
40+
register(new OracleDsToSpSourceConnector());
4241
}
4342

44-
private static void register(ISourceConnector connector) {
43+
private static void register(IDsToSpSourceConnector connector) {
4544
String type = connector.getSourceType().toLowerCase();
4645
if (CONNECTORS.containsKey(type)) {
4746
LOG.warn("Duplicate connector registered for type: {}. Overwriting.", type);
@@ -50,14 +49,14 @@ private static void register(ISourceConnector connector) {
5049
LOG.info("Registered Datastream source connector: {}", type);
5150
}
5251

53-
private SourceConnectorRegistry() {}
52+
private DatastreamToSpannerSourceConnectorRegistry() {}
5453

5554
/** Returns the connector for the given source type, or throws if not found. */
56-
public static ISourceConnector getSourceConnector(String sourceType) {
55+
public static IDsToSpSourceConnector getSourceConnector(String sourceType) {
5756
if (sourceType == null || sourceType.isEmpty()) {
5857
throw new IllegalArgumentException("Source type cannot be empty");
5958
}
60-
ISourceConnector connector = CONNECTORS.get(sourceType.toLowerCase());
59+
IDsToSpSourceConnector connector = CONNECTORS.get(sourceType.toLowerCase());
6160
if (connector == null) {
6261
throw new IllegalArgumentException(
6362
"Unsupported source type: " + sourceType + ". Registered types: " + CONNECTORS.keySet());
@@ -66,7 +65,7 @@ public static ISourceConnector getSourceConnector(String sourceType) {
6665
}
6766

6867
/** Returns all registered connectors. */
69-
public static Collection<ISourceConnector> getConnectors() {
68+
public static Collection<IDsToSpSourceConnector> getConnectors() {
7069
return Collections.unmodifiableCollection(CONNECTORS.values());
7170
}
7271

@@ -77,7 +76,7 @@ public static Set<String> getSupportedSourceTypes() {
7776

7877
/** Identifies the source type from Datastream SourceConfig by querying all connectors. */
7978
public static String getSourceTypeFromConfig(SourceConfig sourceConfig) {
80-
for (ISourceConnector connector : CONNECTORS.values()) {
79+
for (IDsToSpSourceConnector connector : CONNECTORS.values()) {
8180
if (connector.matches(sourceConfig)) {
8281
return connector.getSourceType();
8382
}

v2/datastream-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/datastream/source/ISourceConnector.java renamed to v2/datastream-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/datastream/source/IDsToSpSourceConnector.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@
3333
* Interface for Datastream source connectors. Encapsulates all source-specific logic for processing
3434
* change events and managing sequences.
3535
*/
36-
public interface ISourceConnector {
36+
public interface IDsToSpSourceConnector {
3737

3838
/** Returns the unique identifier for this source type (e.g., "mysql", "postgresql", "oracle"). */
3939
String getSourceType();

v2/datastream-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/datastream/source/mysql/MySqlSourceConnector.java renamed to v2/datastream-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/datastream/source/mysql/MySqlDsToSpSourceConnector.java

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -27,12 +27,12 @@
2727
import com.google.cloud.teleport.v2.templates.datastream.ChangeEventSequence;
2828
import com.google.cloud.teleport.v2.templates.datastream.ChangeEventSequenceCreationException;
2929
import com.google.cloud.teleport.v2.templates.datastream.DatastreamConstants;
30-
import com.google.cloud.teleport.v2.templates.datastream.source.ISourceConnector;
30+
import com.google.cloud.teleport.v2.templates.datastream.source.IDsToSpSourceConnector;
3131
import java.util.Map;
3232
import org.apache.commons.lang3.tuple.Pair;
3333

34-
/** MySQL implementation of {@link ISourceConnector} connector. */
35-
public class MySqlSourceConnector implements ISourceConnector {
34+
/** MySQL implementation of {@link IDsToSpSourceConnector} connector. */
35+
public class MySqlDsToSpSourceConnector implements IDsToSpSourceConnector {
3636

3737
@Override
3838
public String getSourceType() {

v2/datastream-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/datastream/source/oracle/OracleSourceConnector.java renamed to v2/datastream-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/datastream/source/oracle/OracleDsToSpSourceConnector.java

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -27,12 +27,12 @@
2727
import com.google.cloud.teleport.v2.templates.datastream.ChangeEventSequence;
2828
import com.google.cloud.teleport.v2.templates.datastream.ChangeEventSequenceCreationException;
2929
import com.google.cloud.teleport.v2.templates.datastream.DatastreamConstants;
30-
import com.google.cloud.teleport.v2.templates.datastream.source.ISourceConnector;
30+
import com.google.cloud.teleport.v2.templates.datastream.source.IDsToSpSourceConnector;
3131
import java.util.Map;
3232
import org.apache.commons.lang3.tuple.Pair;
3333

34-
/** Oracle implementation of {@link ISourceConnector} connector. */
35-
public class OracleSourceConnector implements ISourceConnector {
34+
/** Oracle implementation of {@link IDsToSpSourceConnector} connector. */
35+
public class OracleDsToSpSourceConnector implements IDsToSpSourceConnector {
3636

3737
@Override
3838
public String getSourceType() {

v2/datastream-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/datastream/source/postgresql/PostgresqlSourceConnector.java renamed to v2/datastream-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/datastream/source/postgresql/PostgresqlDsToSpSourceConnector.java

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -27,12 +27,12 @@
2727
import com.google.cloud.teleport.v2.templates.datastream.ChangeEventSequence;
2828
import com.google.cloud.teleport.v2.templates.datastream.ChangeEventSequenceCreationException;
2929
import com.google.cloud.teleport.v2.templates.datastream.DatastreamConstants;
30-
import com.google.cloud.teleport.v2.templates.datastream.source.ISourceConnector;
30+
import com.google.cloud.teleport.v2.templates.datastream.source.IDsToSpSourceConnector;
3131
import java.util.Map;
3232
import org.apache.commons.lang3.tuple.Pair;
3333

34-
/** PostgreSQL implementation of {@link ISourceConnector} connector. */
35-
public class PostgresqlSourceConnector implements ISourceConnector {
34+
/** PostgreSQL implementation of {@link IDsToSpSourceConnector} connector. */
35+
public class PostgresqlDsToSpSourceConnector implements IDsToSpSourceConnector {
3636

3737
@Override
3838
public String getSourceType() {

‎v2/datastream-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/spanner/ShadowTableCreator.java‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -20,8 +20,8 @@
2020
import com.google.cloud.teleport.v2.spanner.ddl.Ddl;
2121
import com.google.cloud.teleport.v2.spanner.ddl.IndexColumn;
2222
import com.google.cloud.teleport.v2.spanner.ddl.Table;
23-
import com.google.cloud.teleport.v2.templates.datastream.source.ISourceConnector;
24-
import com.google.cloud.teleport.v2.templates.datastream.source.SourceConnectorRegistry;
23+
import com.google.cloud.teleport.v2.templates.datastream.source.DatastreamToSpannerSourceConnectorRegistry;
24+
import com.google.cloud.teleport.v2.templates.datastream.source.IDsToSpSourceConnector;
2525
import java.util.List;
2626
import java.util.Map;
2727
import java.util.Set;
@@ -38,7 +38,8 @@ public class ShadowTableCreator {
3838
ShadowTableCreator(String sourceType, String shadowTablePrefix, Dialect dialect) {
3939
this.sourceType = sourceType;
4040
this.shadowTablePrefix = shadowTablePrefix;
41-
ISourceConnector connector = SourceConnectorRegistry.getSourceConnector(sourceType);
41+
IDsToSpSourceConnector connector =
42+
DatastreamToSpannerSourceConnectorRegistry.getSourceConnector(sourceType);
4243
this.sortOrderMap = connector.getSortOrder(dialect);
4344
}
4445

v2/datastream-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/datastream/source/SourceConnectorRegistryTest.java renamed to v2/datastream-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/datastream/source/DatastreamToSpannerSourceConnectorRegistryTest.java

Lines changed: 24 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -24,18 +24,19 @@
2424
import com.google.api.services.datastream.v1.model.OracleSourceConfig;
2525
import com.google.api.services.datastream.v1.model.PostgresqlSourceConfig;
2626
import com.google.api.services.datastream.v1.model.SourceConfig;
27-
import com.google.cloud.teleport.v2.templates.datastream.source.mysql.MySqlSourceConnector;
28-
import com.google.cloud.teleport.v2.templates.datastream.source.oracle.OracleSourceConnector;
29-
import com.google.cloud.teleport.v2.templates.datastream.source.postgresql.PostgresqlSourceConnector;
27+
import com.google.cloud.teleport.v2.templates.datastream.source.mysql.MySqlDsToSpSourceConnector;
28+
import com.google.cloud.teleport.v2.templates.datastream.source.oracle.OracleDsToSpSourceConnector;
29+
import com.google.cloud.teleport.v2.templates.datastream.source.postgresql.PostgresqlDsToSpSourceConnector;
3030
import java.util.Set;
3131
import org.junit.Test;
3232

33-
/** Tests for {@link SourceConnectorRegistry}. */
34-
public class SourceConnectorRegistryTest {
33+
/** Tests for {@link DatastreamToSpannerSourceConnectorRegistry}. */
34+
public class DatastreamToSpannerSourceConnectorRegistryTest {
3535

3636
@Test
3737
public void testGetSupportedSourceTypes() {
38-
Set<String> supportedTypes = SourceConnectorRegistry.getSupportedSourceTypes();
38+
Set<String> supportedTypes =
39+
DatastreamToSpannerSourceConnectorRegistry.getSupportedSourceTypes();
3940
assertEquals(3, supportedTypes.size());
4041
assertTrue(supportedTypes.contains("mysql"));
4142
assertTrue(supportedTypes.contains("postgresql"));
@@ -45,51 +46,54 @@ public void testGetSupportedSourceTypes() {
4546
@Test
4647
public void testGetSourceConnector() {
4748
assertThat(
48-
SourceConnectorRegistry.getSourceConnector("mysql"),
49-
instanceOf(MySqlSourceConnector.class));
49+
DatastreamToSpannerSourceConnectorRegistry.getSourceConnector("mysql"),
50+
instanceOf(MySqlDsToSpSourceConnector.class));
5051
assertThat(
51-
SourceConnectorRegistry.getSourceConnector("MYSQL"),
52-
instanceOf(MySqlSourceConnector.class));
52+
DatastreamToSpannerSourceConnectorRegistry.getSourceConnector("MYSQL"),
53+
instanceOf(MySqlDsToSpSourceConnector.class));
5354
assertThat(
54-
SourceConnectorRegistry.getSourceConnector("postgresql"),
55-
instanceOf(PostgresqlSourceConnector.class));
55+
DatastreamToSpannerSourceConnectorRegistry.getSourceConnector("postgresql"),
56+
instanceOf(PostgresqlDsToSpSourceConnector.class));
5657
assertThat(
57-
SourceConnectorRegistry.getSourceConnector("oracle"),
58-
instanceOf(OracleSourceConnector.class));
58+
DatastreamToSpannerSourceConnectorRegistry.getSourceConnector("oracle"),
59+
instanceOf(OracleDsToSpSourceConnector.class));
5960
}
6061

6162
@Test(expected = IllegalArgumentException.class)
6263
public void testGetSourceConnectorNotFound() {
63-
SourceConnectorRegistry.getSourceConnector("invalid");
64+
DatastreamToSpannerSourceConnectorRegistry.getSourceConnector("invalid");
6465
}
6566

6667
@Test(expected = IllegalArgumentException.class)
6768
public void testGetSourceConnectorEmpty() {
68-
SourceConnectorRegistry.getSourceConnector("");
69+
DatastreamToSpannerSourceConnectorRegistry.getSourceConnector("");
6970
}
7071

7172
@Test
7273
public void testGetSourceTypeFromConfigMysql() {
7374
SourceConfig config = new SourceConfig().setMysqlSourceConfig(new MysqlSourceConfig());
74-
assertEquals("mysql", SourceConnectorRegistry.getSourceTypeFromConfig(config));
75+
assertEquals(
76+
"mysql", DatastreamToSpannerSourceConnectorRegistry.getSourceTypeFromConfig(config));
7577
}
7678

7779
@Test
7880
public void testGetSourceTypeFromConfigPostgresql() {
7981
SourceConfig config =
8082
new SourceConfig().setPostgresqlSourceConfig(new PostgresqlSourceConfig());
81-
assertEquals("postgresql", SourceConnectorRegistry.getSourceTypeFromConfig(config));
83+
assertEquals(
84+
"postgresql", DatastreamToSpannerSourceConnectorRegistry.getSourceTypeFromConfig(config));
8285
}
8386

8487
@Test
8588
public void testGetSourceTypeFromConfigOracle() {
8689
SourceConfig config = new SourceConfig().setOracleSourceConfig(new OracleSourceConfig());
87-
assertEquals("oracle", SourceConnectorRegistry.getSourceTypeFromConfig(config));
90+
assertEquals(
91+
"oracle", DatastreamToSpannerSourceConnectorRegistry.getSourceTypeFromConfig(config));
8892
}
8993

9094
@Test(expected = IllegalArgumentException.class)
9195
public void testGetSourceTypeFromConfigUnsupported() {
9296
SourceConfig config = new SourceConfig();
93-
SourceConnectorRegistry.getSourceTypeFromConfig(config);
97+
DatastreamToSpannerSourceConnectorRegistry.getSourceTypeFromConfig(config);
9498
}
9599
}

‎v2/datastream-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/datastream/source/mysql/MySqlChangeEventContextTest.java‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -73,7 +73,7 @@ public void canGenerateShadowTableMutation() throws Exception {
7373
DatastreamConstants.EVENT_SOURCE_TYPE_KEY, DatastreamConstants.MYSQL_SOURCE_TYPE);
7474

7575
ChangeEventContext changeEventContext =
76-
new MySqlSourceConnector()
76+
new MySqlDsToSpSourceConnector()
7777
.createChangeEventContext(getJsonNode(changeEvent.toString()), ddl, ddl, "shadow_");
7878
Mutation shadowMutation = changeEventContext.getShadowTableMutation();
7979
Map<String, Value> actual = shadowMutation.asMap();
@@ -112,7 +112,7 @@ public void canGenerateShadowTableMutationForBackfillEvents() throws Exception {
112112
DatastreamConstants.EVENT_SOURCE_TYPE_KEY, DatastreamConstants.MYSQL_SOURCE_TYPE);
113113

114114
ChangeEventContext changeEventContext =
115-
new MySqlSourceConnector()
115+
new MySqlDsToSpSourceConnector()
116116
.createChangeEventContext(getJsonNode(changeEvent.toString()), ddl, ddl, "shadow_");
117117
Mutation shadowMutation = changeEventContext.getShadowTableMutation();
118118
Map<String, Value> actual = shadowMutation.asMap();
@@ -149,7 +149,7 @@ public void canGenerateShadowTableMutationForBackfillEventsWithMissingSortOrderK
149149
DatastreamConstants.EVENT_SOURCE_TYPE_KEY, DatastreamConstants.MYSQL_SOURCE_TYPE);
150150

151151
ChangeEventContext changeEventContext =
152-
new MySqlSourceConnector()
152+
new MySqlDsToSpSourceConnector()
153153
.createChangeEventContext(getJsonNode(changeEvent.toString()), ddl, ddl, "shadow_");
154154
Mutation shadowMutation = changeEventContext.getShadowTableMutation();
155155
Map<String, Value> actual = shadowMutation.asMap();
@@ -219,7 +219,7 @@ public void canGenerateShadowTableMutationWithCollision() throws Exception {
219219
changeEvent.put(DatastreamConstants.MYSQL_LOGPOSITION_KEY, 100L);
220220

221221
ChangeEventContext changeEventContext =
222-
new MySqlSourceConnector()
222+
new MySqlDsToSpSourceConnector()
223223
.createChangeEventContext(
224224
getJsonNode(changeEvent.toString()), ddl, shadowDdl, "shadow_");
225225
Mutation shadowMutation = changeEventContext.getShadowTableMutation();
@@ -244,7 +244,7 @@ public void testReadDataTable() throws Exception {
244244
DatastreamConstants.EVENT_SOURCE_TYPE_KEY, DatastreamConstants.MYSQL_SOURCE_TYPE);
245245

246246
ChangeEventContext context =
247-
new MySqlSourceConnector()
247+
new MySqlDsToSpSourceConnector()
248248
.createChangeEventContext(getJsonNode(changeEvent.toString()), ddl, ddl, "shadow_");
249249

250250
TransactionContext transactionContext = mock(TransactionContext.class);
@@ -335,7 +335,7 @@ public void testCanGenerateDataDmlStatement() throws Exception {
335335
changeEvent.put("id", "aa");
336336

337337
ChangeEventContext context =
338-
new MySqlSourceConnector()
338+
new MySqlDsToSpSourceConnector()
339339
.createChangeEventContext(getJsonNode(changeEvent.toString()), ddl, ddl, "shadow_");
340340

341341
Statement dmlStatement = context.getDataDmlStatement(ddl);

0 commit comments

Comments
 (0)