Add the new `change.stream.full.document.before.change` config property #121

stIncMale · 2022-10-20T23:19:45Z

This new property is equivalent to ChangeStreamIterable.fullDocumentBeforeChange in the driver API.

Problems:

~~4 tests in MongoSourceTaskTest accidentally became integration tests (they need a running MongoDB). These tests need to be refactored.~~
- Solution: rename MongoSourceTaskTest to MongoSourceTaskIntegrationJunkTest and move it to integration tests. These tests are irredeemable and need to be rewritten to become unit tests.

…nt.before.change` config property KAFKA-308

stIncMale · 2022-10-20T23:24:06Z

src/main/java/com/mongodb/kafka/connect/source/MongoSourceTask.java

  }

  @Override
-  @SuppressWarnings("try")
-  public synchronized void stop() {


Removed synchronized as stop cannot be called concurrently with itself, only with poll/commit, but poll/commit were neither synchronized nor used synchronized (this) blocks.

stIncMale · 2022-10-20T23:25:21Z

src/main/java/com/mongodb/kafka/connect/source/StartedMongoSourceTask.java

+  private final Supplier<SourceTaskContext> sourceTaskContextAccessor;
+  private final Time time;
+  private volatile boolean isRunning;
+  private boolean isCopying;


isCopying is now a plain field because there is no point in setting it to false in stop. And without writes from stop, this field is always accessed sequentially.

stIncMale · 2022-10-20T23:30:50Z

src/test/java/com/mongodb/kafka/connect/source/producer/SchemaAndValueProducerTest.java

@@ -280,7 +280,7 @@ void testBsonSchemaAndValueProducer() {
        "Assert schema and value matches",
        () -> assertEquals(Schema.BYTES_SCHEMA.schema(), actual.schema()),
        // Ensure the data length is truncated.
-        () -> assertEquals(1071, ((byte[]) actual.value()).length),
+        () -> assertEquals(1712, ((byte[]) actual.value()).length),


@rozza I have no idea where the previous 1071 value came from, or what this assertion is. I put 1712 as this is the new value we have in this test as a result of changes. Should we just delete the assertion as it is useless, given that a reader cannot understand what this is, and doing what I did appears to be the only way to make schema changes, unless the reader has access to arcane knowledge?

Did you expect this value to change as a result of your changes? At the very least, it tests against an unexpected change to this value, it seems better to fix the comment (I am not sure how it tests that the length is truncated) than to delete.

It is expected that the serialized value is longer, as it has more data now. But I know neither where the asserted value is supposed to come from, nor why the serialized value is expected to be truncated.

We should not delete the assertion, because it guards against unexpected changes to the value, even though we do not know exactly how to compute the value.

iirc it was added to ensure that the size is as expected - it really is testing documentToByteArray and its handling of RawBsonDocuments. So I agree it can be removed from here and is better suited to its own test and / or is already tested in BsonValueToSchemaAndValueTest

Removed the assertions.

KAFKA-308

stIncMale · 2022-10-24T17:12:34Z

src/main/java/com/mongodb/kafka/connect/source/StartedMongoSourceTask.java

+  StartedMongoSourceTask(
+      final Supplier<SourceTaskContext> sourceTaskContextAccessor,
+      final MongoSourceConfig sourceConfig,
+      final MongoClient mongoClient,
+      @Nullable final MongoCopyDataManager copyDataManager,
+      final StatisticsManager statisticsManager) {


StartedMongoSourceTask was extracted from MongoSourceTask, it gets all its dependencies via a constructor, which makes it unit-testable.

stIncMale · 2022-10-24T17:23:52Z

src/test/java/com/mongodb/kafka/connect/source/StartedMongoSourceTaskTest.java

+final class StartedMongoSourceTaskTest {
+  static final class FullDocumentBeforeChangeTest {


I am proposing the idea of grouping unit tests in different static nested classes by functionality, as this prevents cluttering the test state, set up and tear down methods. If you think this is something that should not be done, or that it may be done, but only when we have unit-tests for different parts of functionality, or something else, please express your thoughts.

Anything that validates the connector and makes the code easier to read / come back to / consume sounds great to me. Do you want to add a separate ticket for this?

I don't think we need to go and update existing tests, but rather I was curious whether it is fine to start using the approach when writing new tests (and the answer appears to be "yes"), if that seems helpful. As for the existing tests, we may regroup them when we touch them if, again, that seems beneficial in a specific case.

stIncMale · 2022-10-24T18:56:15Z

src/main/java/com/mongodb/kafka/connect/source/MongoSourceConfig.java

+  private static final String FULL_DOCUMENT_BEFORE_CHANGE_DOC =
+      "Specifies the pre-image configuration when creating a Change Stream.\n"
+          + "The pre-image is not available in source records published while copying existing data as a result of"
+          + " enabling `copy.existing`, and the pre-image configuration has no effect on copying.\n"
+          + "See https://www.mongodb.com/docs/manual/reference/method/db.collection.watch/ for more details and possible values.";


I linked the server docs here instead of listing and explaining all possible values, because, as always, duplicating the docs eventually leads to the duplicated docs becoming inconsistent and/or obsolete, which results in misleading users, which in turn, makes users to distrust the docs.

In the past, I gave examples of problems our approach introduced in the driver API (where the decision has been and stays "the convenience of reading docs in place is more important than having consistent docs users may trust"), and we have similar situations in Kafka. For example, the documentation for change.stream.full.document config property states that possible values are only "", "default", "updateLookup". The new values "whenAvailable" and "required" are omitted, as they were added later - what should users think about this? Does the connector not support these values? Does it support them, but the authors forgot to specify them? If users actually trust the connector docs (they should not), they will not learn about these two new values, or any values that will be supported by the server in the future.

KAFKA-308

katcharov

Looking good

katcharov · 2022-10-24T22:21:04Z

src/test/java/com/mongodb/kafka/connect/source/producer/SchemaAndValueProducerTest.java

@@ -280,7 +280,7 @@ void testBsonSchemaAndValueProducer() {
        "Assert schema and value matches",
        () -> assertEquals(Schema.BYTES_SCHEMA.schema(), actual.schema()),
        // Ensure the data length is truncated.
-        () -> assertEquals(1071, ((byte[]) actual.value()).length),
+        () -> assertEquals(1712, ((byte[]) actual.value()).length),


Did you expect this value to change as a result of your changes? At the very least, it tests against an unexpected change to this value, it seems better to fix the comment (I am not sure how it tests that the length is truncated) than to delete.

...ntegrationTest/java/com/mongodb/kafka/connect/source/MongoSourceTaskIntegrationJunkTest.java

src/main/java/com/mongodb/kafka/connect/sink/StartedMongoSinkTask.java

src/main/java/com/mongodb/kafka/connect/source/statistics/JmxStatisticsManager.java

…ationTest2` KAFKA-308

rozza

One naming suggestion but LGTM :)

rozza · 2022-10-31T15:27:45Z

src/integrationTest/java/com/mongodb/kafka/connect/source/MongoSourceTaskIntegrationTest2.java

 @ExtendWith(MockitoExtension.class)
-class MongoSourceTaskTest {
+class MongoSourceTaskIntegrationTest2 {


Suggestion name: MongoSourceTaskVerificationTest

What does it mean for a test to be verification test? It appears to me that all tests verify something. I'd prefer to leave the ...IntegrationTest2 name. It won't survive past KAFKA-234.

rozza · 2022-10-31T15:28:20Z

src/main/java/com/mongodb/kafka/connect/source/MongoSourceConfig.java

+  private static final String FULL_DOCUMENT_BEFORE_CHANGE_DOC =
+      "Specifies the pre-image configuration when creating a Change Stream.\n"
+          + "The pre-image is not available in source records published while copying existing data as a result of"
+          + " enabling `copy.existing`, and the pre-image configuration has no effect on copying.\n"
+          + "See https://www.mongodb.com/docs/manual/reference/method/db.collection.watch/ for more details and possible values.";


rozza · 2022-10-31T15:29:15Z

src/main/java/com/mongodb/kafka/connect/source/MongoSourceTask.java

  }

  @Override
-  @SuppressWarnings("try")
-  public synchronized void stop() {


rozza · 2022-10-31T15:32:58Z

src/test/java/com/mongodb/kafka/connect/source/StartedMongoSourceTaskTest.java

+final class StartedMongoSourceTaskTest {
+  static final class FullDocumentBeforeChangeTest {


Anything that validates the connector and makes the code easier to read / come back to / consume sounds great to me. Do you want to add a separate ticket for this?

rozza · 2022-10-31T15:45:31Z

src/test/java/com/mongodb/kafka/connect/source/producer/SchemaAndValueProducerTest.java

@@ -280,7 +280,7 @@ void testBsonSchemaAndValueProducer() {
        "Assert schema and value matches",
        () -> assertEquals(Schema.BYTES_SCHEMA.schema(), actual.schema()),
        // Ensure the data length is truncated.
-        () -> assertEquals(1071, ((byte[]) actual.value()).length),
+        () -> assertEquals(1712, ((byte[]) actual.value()).length),


iirc it was added to ensure that the size is as expected - it really is testing documentToByteArray and its handling of RawBsonDocuments. So I agree it can be removed from here and is better suited to its own test and / or is already tested in BsonValueToSchemaAndValueTest

KAFKA-308

rozza

LGTM

Update the default Avro schema and add the `change.stream.full.docume…

87c5f0c

…nt.before.change` config property KAFKA-308

stIncMale requested review from katcharov and rozza October 20, 2022 23:19

stIncMale self-assigned this Oct 20, 2022

stIncMale commented Oct 20, 2022

View reviewed changes

stIncMale marked this pull request as draft October 20, 2022 23:45

stIncMale force-pushed the KAFKA-308 branch from f318648 to 91bd27e Compare October 21, 2022 17:06

stIncMale marked this pull request as ready for review October 21, 2022 17:06

stIncMale marked this pull request as draft October 21, 2022 17:28

stIncMale force-pushed the KAFKA-308 branch from 91bd27e to cb30719 Compare October 21, 2022 18:55

Refactor: extract StartedMongoSourceTask from MongoSourceTask

23f79f9

KAFKA-308

stIncMale force-pushed the KAFKA-308 branch from cb30719 to 23f79f9 Compare October 24, 2022 15:30

Handle MongoSourceConfig.getFullDocumentBeforeChange and add tests

23ab6e1

KAFKA-308

stIncMale commented Oct 24, 2022

View reviewed changes

stIncMale marked this pull request as ready for review October 24, 2022 17:30

stIncMale requested review from katcharov and rozza and removed request for rozza and katcharov October 24, 2022 17:30

stIncMale commented Oct 24, 2022

View reviewed changes

stIncMale added 2 commits October 24, 2022 13:06

Update MongoSourceConfig.FULL_DOCUMENT_DOC

a5e6bca

KAFKA-308

Put the lost comments back

c9f4bba

KAFKA-308

katcharov reviewed Oct 25, 2022

View reviewed changes

stIncMale requested a review from katcharov October 25, 2022 20:14

katcharov approved these changes Oct 27, 2022

View reviewed changes

Rename MongoSourceTaskIntegrationJunkTest to `MongoSourceTaskIntegr…

bdb11ff

…ationTest2` KAFKA-308

rozza approved these changes Oct 31, 2022

View reviewed changes

stIncMale added 2 commits October 31, 2022 11:36

Improve the docs

0af3144

KAFKA-308

Remove arcane assertions from SchemaAndValueProducerTest

8011b86

KAFKA-308

stIncMale requested a review from rozza October 31, 2022 17:45

rozza approved these changes Nov 1, 2022

View reviewed changes

stIncMale merged commit 422f53b into mongodb:master Nov 1, 2022

stIncMale deleted the KAFKA-308 branch November 1, 2022 14:44

stIncMale mentioned this pull request Oct 21, 2023

Preserve error code, code name, and error labels when redacting command monitoring/logging mongodb/mongo-java-driver#1225

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add the new `change.stream.full.document.before.change` config property #121

Add the new `change.stream.full.document.before.change` config property #121

stIncMale commented Oct 20, 2022 •

edited

stIncMale Oct 20, 2022

rozza Oct 31, 2022

stIncMale Oct 20, 2022

stIncMale Oct 20, 2022

katcharov Oct 24, 2022

stIncMale Oct 25, 2022

katcharov Oct 26, 2022

rozza Oct 31, 2022

stIncMale Oct 31, 2022

stIncMale Oct 24, 2022

stIncMale Oct 24, 2022

rozza Oct 31, 2022

stIncMale Oct 31, 2022

stIncMale Oct 24, 2022 •

edited

rozza Oct 31, 2022

katcharov left a comment

katcharov Oct 24, 2022

rozza left a comment

rozza Oct 31, 2022

stIncMale Oct 31, 2022

rozza Oct 31, 2022

rozza Oct 31, 2022

rozza Oct 31, 2022

rozza Oct 31, 2022

rozza left a comment

		final class StartedMongoSourceTaskTest {
		static final class FullDocumentBeforeChangeTest {

Add the new change.stream.full.document.before.change config property #121

Add the new change.stream.full.document.before.change config property #121

Conversation

stIncMale commented Oct 20, 2022 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

stIncMale Oct 24, 2022 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

katcharov left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rozza left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rozza left a comment

Choose a reason for hiding this comment

Add the new `change.stream.full.document.before.change` config property #121

Add the new `change.stream.full.document.before.change` config property #121

stIncMale commented Oct 20, 2022 •

edited

stIncMale Oct 24, 2022 •

edited