Skip to content

Commit bbd6aaa

Browse files
hpgrahslhpgrahsl
authored
Merge pull request #22 from hpgrahsl/code-quality-test-coverage
code quality improvements
2 parents f086ef8 + 60b4a00 commit bbd6aaa

File tree

13 files changed

+383
-56
lines changed

13 files changed

+383
-56
lines changed

‎pom.xml‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -10,13 +10,13 @@
1010
<name>kafka-connect-mongodb</name>
1111

1212
<properties>
13-
<kafka.version>1.0.0</kafka.version>
13+
<kafka.version>1.0.1</kafka.version>
1414
<junit.version>4.12</junit.version>
15-
<mongodb.driver.version>3.6.0</mongodb.driver.version>
15+
<mongodb.driver.version>3.6.3</mongodb.driver.version>
1616
<logback.version>1.2.3</logback.version>
17-
<junit.jupiter.version>5.0.2</junit.jupiter.version>
18-
<junit.vintage.version>4.12.2</junit.vintage.version>
19-
<junit.platform.version>1.0.2</junit.platform.version>
17+
<junit.jupiter.version>5.1.0</junit.jupiter.version>
18+
<junit.vintage.version>5.1.0</junit.vintage.version>
19+
<junit.platform.version>1.1.0</junit.platform.version>
2020
<mockito.version>2.9.0</mockito.version>
2121
<hamcrest.version>2.0.0.0</hamcrest.version>
2222
<jackson.version>2.9.0</jackson.version>

‎src/main/java/at/grahsl/kafka/connect/mongodb/cdc/debezium/mongodb/MongoDbHandler.java‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,18 +35,24 @@ public class MongoDbHandler extends DebeziumCdcHandler {
3535

3636
public static final String JSON_ID_FIELD_PATH = "id";
3737

38-
private final Map<OperationType,CdcOperation> operations = new HashMap<>();
3938
private static Logger logger = LoggerFactory.getLogger(MongoDbHandler.class);
4039

4140
public MongoDbHandler(MongoDbSinkConnectorConfig config) {
4241
super(config);
42+
final Map<OperationType,CdcOperation> operations = new HashMap<>();
4343
operations.put(OperationType.CREATE,new MongoDbInsert());
4444
operations.put(OperationType.READ,new MongoDbInsert());
4545
operations.put(OperationType.UPDATE,new MongoDbUpdate());
4646
operations.put(OperationType.DELETE,new MongoDbDelete());
4747
registerOperations(operations);
4848
}
4949

50+
public MongoDbHandler(MongoDbSinkConnectorConfig config,
51+
Map<OperationType,CdcOperation> operations) {
52+
super(config);
53+
registerOperations(operations);
54+
}
55+
5056
@Override
5157
public Optional<WriteModel<BsonDocument>> handle(SinkDocument doc) {
5258

‎src/main/java/at/grahsl/kafka/connect/mongodb/cdc/debezium/mysql/MysqlHandler.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,12 @@ public MysqlHandler(MongoDbSinkConnectorConfig config) {
5151
registerOperations(operations);
5252
}
5353

54+
public MysqlHandler(MongoDbSinkConnectorConfig config,
55+
Map<OperationType,CdcOperation> operations) {
56+
super(config);
57+
registerOperations(operations);
58+
}
59+
5460
@Override
5561
public Optional<WriteModel<BsonDocument>> handle(SinkDocument doc) {
5662

‎src/main/java/at/grahsl/kafka/connect/mongodb/converter/types/sink/bson/BytesFieldConverter.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -41,8 +41,8 @@ public BsonValue toBson(Object data) {
4141
if(data instanceof byte[])
4242
return new BsonBinary((byte[])data);
4343

44-
throw new DataException("error: bytes field conversion failed to due "
45-
+ "to unexpected object type "+ data.getClass().getName());
44+
throw new DataException("error: bytes field conversion failed to due"
45+
+ " unexpected object type "+ data.getClass().getName());
4646

4747
}
4848

‎src/main/java/at/grahsl/kafka/connect/mongodb/converter/types/sink/bson/logical/DecimalFieldConverter.java‎

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -48,14 +48,16 @@ public DecimalFieldConverter(Format format) {
4848
@Override
4949
public BsonValue toBson(Object data) {
5050

51-
if(format.equals(Format.DECIMAL128))
52-
return new BsonDecimal128(new Decimal128((BigDecimal)data));
51+
if(data instanceof BigDecimal) {
52+
if(format.equals(Format.DECIMAL128))
53+
return new BsonDecimal128(new Decimal128((BigDecimal)data));
5354

54-
if(format.equals(Format.LEGACYDOUBLE))
55-
return new BsonDouble(((BigDecimal)data).doubleValue());
55+
if(format.equals(Format.LEGACYDOUBLE))
56+
return new BsonDouble(((BigDecimal)data).doubleValue());
57+
}
5658

57-
throw new DataException("error: decimal conversion using format "
58-
+ format + " not supported");
59+
throw new DataException("error: decimal conversion not possible when data is"
60+
+ " of type "+data.getClass().getName() + " and format is "+format);
5961

6062
}
6163
}

‎src/main/java/at/grahsl/kafka/connect/mongodb/processor/DocumentIdAdder.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,7 @@
2424

2525
public class DocumentIdAdder extends PostProcessor {
2626

27-
IdStrategy idStrategy;
27+
protected final IdStrategy idStrategy;
2828

2929
public DocumentIdAdder(MongoDbSinkConnectorConfig config) {
3030
this(config,config.getIdStrategy());

‎src/test/java/at/grahsl/kafka/connect/mongodb/cdc/debezium/OperationTypeTest.java‎

Lines changed: 25 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -5,34 +5,53 @@
55
import org.junit.platform.runner.JUnitPlatform;
66
import org.junit.runner.RunWith;
77

8-
import static org.junit.jupiter.api.Assertions.assertEquals;
9-
import static org.junit.jupiter.api.Assertions.assertThrows;
8+
import static org.junit.jupiter.api.Assertions.*;
109

1110
@RunWith(JUnitPlatform.class)
1211
public class OperationTypeTest {
1312

1413
@Test
1514
@DisplayName("when op type 'c' then type CREATE")
1615
public void testOperationTypeCreate() {
17-
assertEquals(OperationType.CREATE,OperationType.fromText("c"));
16+
String textType = "c";
17+
OperationType otCreate = OperationType.fromText(textType);
18+
assertAll(
19+
() -> assertEquals(OperationType.CREATE,otCreate),
20+
() -> assertEquals(textType,otCreate.type())
21+
);
1822
}
1923

2024
@Test
2125
@DisplayName("when op type 'r' then type READ")
2226
public void testOperationTypeRead() {
23-
assertEquals(OperationType.READ,OperationType.fromText("r"));
27+
String textType = "r";
28+
OperationType otRead = OperationType.fromText(textType);
29+
assertAll(
30+
() -> assertEquals(OperationType.READ,otRead),
31+
() -> assertEquals(textType,otRead.type())
32+
);
2433
}
2534

2635
@Test
2736
@DisplayName("when op type 'u' then type UPDATE")
2837
public void testOperationTypeUpdate() {
29-
assertEquals(OperationType.UPDATE,OperationType.fromText("u"));
38+
String textType = "u";
39+
OperationType otUpdate = OperationType.fromText(textType);
40+
assertAll(
41+
() -> assertEquals(OperationType.UPDATE,otUpdate),
42+
() -> assertEquals(textType,otUpdate.type())
43+
);
3044
}
3145

3246
@Test
3347
@DisplayName("when op type 'd' then type DELETE")
3448
public void testOperationTypeDelete() {
35-
assertEquals(OperationType.DELETE,OperationType.fromText("d"));
49+
String textType = "d";
50+
OperationType otDelete = OperationType.fromText(textType);
51+
assertAll(
52+
() -> assertEquals(OperationType.DELETE,otDelete),
53+
() -> assertEquals(textType,otDelete.type())
54+
);
3655
}
3756

3857
@Test

‎src/test/java/at/grahsl/kafka/connect/mongodb/cdc/debezium/mongodb/MongoDbHandlerTest.java‎

Lines changed: 111 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,10 @@
33
import at.grahsl.kafka.connect.mongodb.MongoDbSinkConnectorConfig;
44
import at.grahsl.kafka.connect.mongodb.cdc.debezium.OperationType;
55
import at.grahsl.kafka.connect.mongodb.converter.SinkDocument;
6+
import com.mongodb.client.model.DeleteOneModel;
7+
import com.mongodb.client.model.ReplaceOneModel;
8+
import com.mongodb.client.model.UpdateOneModel;
9+
import com.mongodb.client.model.WriteModel;
610
import org.apache.kafka.connect.errors.DataException;
711
import org.bson.BsonDocument;
812
import org.bson.BsonInt32;
@@ -25,22 +29,37 @@
2529
@RunWith(JUnitPlatform.class)
2630
public class MongoDbHandlerTest {
2731

28-
public static final MongoDbHandler MONGODB_HANDLER =
32+
public static final MongoDbHandler MONGODB_HANDLER_DEFAULT_MAPPING =
2933
new MongoDbHandler(new MongoDbSinkConnectorConfig(new HashMap<>()));
3034

35+
public static final MongoDbHandler MONGODB_HANDLER_EMPTY_MAPPING =
36+
new MongoDbHandler(new MongoDbSinkConnectorConfig(new HashMap<>()),
37+
new HashMap<>());
38+
39+
@Test
40+
@DisplayName("verify existing default config from base class")
41+
public void testExistingDefaultConfig() {
42+
assertAll(
43+
() -> assertNotNull(MONGODB_HANDLER_DEFAULT_MAPPING.getConfig(),
44+
() -> "default config for handler must not be null"),
45+
() -> assertNotNull(MONGODB_HANDLER_EMPTY_MAPPING.getConfig(),
46+
() -> "default config for handler must not be null")
47+
);
48+
}
49+
3150
@Test
3251
@DisplayName("when key document missing then DataException")
3352
public void testMissingKeyDocument() {
3453
assertThrows(DataException.class, () ->
35-
MONGODB_HANDLER.handle(new SinkDocument(null,null))
54+
MONGODB_HANDLER_DEFAULT_MAPPING.handle(new SinkDocument(null,null))
3655
);
3756
}
3857

3958
@Test
4059
@DisplayName("when key doc contains 'id' field but value is empty then null due to tombstone")
4160
public void testTombstoneEvent() {
4261
assertEquals(Optional.empty(),
43-
MONGODB_HANDLER.handle(new SinkDocument(
62+
MONGODB_HANDLER_DEFAULT_MAPPING.handle(new SinkDocument(
4463
new BsonDocument("id",new BsonInt32(1234)),
4564
new BsonDocument())
4665
),
@@ -56,7 +75,20 @@ public void testUnkownCdcOperationType() {
5675
new BsonDocument("op",new BsonString("x"))
5776
);
5877
assertThrows(DataException.class, () ->
59-
MONGODB_HANDLER.handle(cdcEvent)
78+
MONGODB_HANDLER_DEFAULT_MAPPING.handle(cdcEvent)
79+
);
80+
}
81+
82+
@Test
83+
@DisplayName("when value doc contains unmapped operation type then DataException")
84+
public void testUnmappedCdcOperationType() {
85+
SinkDocument cdcEvent = new SinkDocument(
86+
new BsonDocument("_id",new BsonInt32(1234)),
87+
new BsonDocument("op",new BsonString("c"))
88+
.append("after",new BsonString("{_id:1234,foo:\"blah\"}"))
89+
);
90+
assertThrows(DataException.class, () ->
91+
MONGODB_HANDLER_EMPTY_MAPPING.handle(cdcEvent)
6092
);
6193
}
6294

@@ -68,7 +100,7 @@ public void testInvalidCdcOperationType() {
68100
new BsonDocument("op",new BsonInt32('c'))
69101
);
70102
assertThrows(DataException.class, () ->
71-
MONGODB_HANDLER.handle(cdcEvent)
103+
MONGODB_HANDLER_DEFAULT_MAPPING.handle(cdcEvent)
72104
);
73105
}
74106

@@ -80,32 +112,96 @@ public void testMissingCdcOperationType() {
80112
new BsonDocument("po", BsonNull.VALUE)
81113
);
82114
assertThrows(DataException.class, () ->
83-
MONGODB_HANDLER.handle(cdcEvent)
115+
MONGODB_HANDLER_DEFAULT_MAPPING.handle(cdcEvent)
84116
);
85117
}
86118

87119
@TestFactory
88-
@DisplayName("when valid cdc operation type then correct MongoDB CdcOperation")
89-
public Stream<DynamicTest> testValidCdcOpertionTypes() {
120+
@DisplayName("when valid CDC event then correct WriteModel")
121+
public Stream<DynamicTest> testValidCdcDocument() {
122+
123+
return Stream.of(
124+
dynamicTest("test operation "+OperationType.CREATE, () -> {
125+
Optional<WriteModel<BsonDocument>> result =
126+
MONGODB_HANDLER_DEFAULT_MAPPING.handle(
127+
new SinkDocument(
128+
new BsonDocument("_id",new BsonString("1234")),
129+
new BsonDocument("op",new BsonString("c"))
130+
.append("after",new BsonString("{_id:1234,foo:\"blah\"}"))
131+
)
132+
);
133+
assertTrue(result.isPresent());
134+
assertTrue(result.get() instanceof ReplaceOneModel,
135+
() -> "result expected to be of type ReplaceOneModel");
136+
137+
}),
138+
dynamicTest("test operation "+OperationType.READ, () -> {
139+
Optional<WriteModel<BsonDocument>> result =
140+
MONGODB_HANDLER_DEFAULT_MAPPING.handle(
141+
new SinkDocument(
142+
new BsonDocument("_id",new BsonString("1234")),
143+
new BsonDocument("op",new BsonString("r"))
144+
.append("after",new BsonString("{_id:1234,foo:\"blah\"}"))
145+
)
146+
);
147+
assertTrue(result.isPresent());
148+
assertTrue(result.get() instanceof ReplaceOneModel,
149+
() -> "result expected to be of type ReplaceOneModel");
150+
151+
}),
152+
dynamicTest("test operation "+OperationType.UPDATE, () -> {
153+
Optional<WriteModel<BsonDocument>> result =
154+
MONGODB_HANDLER_DEFAULT_MAPPING.handle(
155+
new SinkDocument(
156+
new BsonDocument("id",new BsonString("1234")),
157+
new BsonDocument("op",new BsonString("u"))
158+
.append("patch",new BsonString("{\"$set\":{foo:\"blah\"}}"))
159+
)
160+
);
161+
assertTrue(result.isPresent());
162+
assertTrue(result.get() instanceof UpdateOneModel,
163+
() -> "result expected to be of type UpdateOneModel");
164+
165+
}),
166+
dynamicTest("test operation "+OperationType.DELETE, () -> {
167+
Optional<WriteModel<BsonDocument>> result =
168+
MONGODB_HANDLER_DEFAULT_MAPPING.handle(
169+
new SinkDocument(
170+
new BsonDocument("id",new BsonString("1234")),
171+
new BsonDocument("op",new BsonString("d"))
172+
)
173+
);
174+
assertTrue(result.isPresent(), () -> "write model result must be present");
175+
assertTrue(result.get() instanceof DeleteOneModel,
176+
() -> "result expected to be of type DeleteOneModel");
177+
178+
})
179+
);
180+
181+
}
182+
183+
@TestFactory
184+
@DisplayName("when valid cdc operation type then correct MongoDB CdcOperation")
185+
public Stream<DynamicTest> testValidCdcOpertionTypes() {
90186

91187
return Stream.of(
92188
dynamicTest("test operation "+OperationType.CREATE, () ->
93-
assertTrue(MONGODB_HANDLER.getCdcOperation(
94-
new BsonDocument("op",new BsonString("c")))
95-
instanceof MongoDbInsert)
96-
),
189+
assertTrue(MONGODB_HANDLER_DEFAULT_MAPPING.getCdcOperation(
190+
new BsonDocument("op",new BsonString("c")))
191+
instanceof MongoDbInsert)
192+
),
97193
dynamicTest("test operation "+OperationType.READ, () ->
98-
assertTrue(MONGODB_HANDLER.getCdcOperation(
194+
assertTrue(MONGODB_HANDLER_DEFAULT_MAPPING.getCdcOperation(
99195
new BsonDocument("op",new BsonString("r")))
100196
instanceof MongoDbInsert)
101197
),
102198
dynamicTest("test operation "+OperationType.UPDATE, () ->
103-
assertTrue(MONGODB_HANDLER.getCdcOperation(
199+
assertTrue(MONGODB_HANDLER_DEFAULT_MAPPING.getCdcOperation(
104200
new BsonDocument("op",new BsonString("u")))
105201
instanceof MongoDbUpdate)
106202
),
107203
dynamicTest("test operation "+OperationType.DELETE, () ->
108-
assertTrue(MONGODB_HANDLER.getCdcOperation(
204+
assertTrue(MONGODB_HANDLER_DEFAULT_MAPPING.getCdcOperation(
109205
new BsonDocument("op",new BsonString("d")))
110206
instanceof MongoDbDelete)
111207
)

0 commit comments

Comments
 (0)