[source-MongoDB] incremental sync giving errors

vatsal-kavida commented 6 months ago

Connector Name

source-mongodb

Connector Version

1.3.3

What step the error happened?

During the sync

Relevant information

have deployed my airbyte on eks using helm. I have setup a mongodb source connector and a postgres destination for incremetal updates. When trying tor un the sync it gives errors -> "io.airbyte.commons.exceptions.ConfigErrorException: Saved offset is not valid. Please reset the connection, and then increase oplog retention and/or increase sync frequency to prevent his from happening in the future. See https://docs.airbyte.com/integrations/sources/mongodb-v2#mongodb-oplog-and-change-streams for more details"

Relevant log output

Here is the whole log output for the sync ->
2024-04-11 13:37:47 replication-orchestrator > sync summary: {
  "status" : "failed",
  "startTime" : 1712842649996,
  "endTime" : 1712842667491,
  "totalStats" : {
    "bytesEmitted" : 0,
    "destinationStateMessagesEmitted" : 0,
    "destinationWriteEndTime" : 1712842667344,
    "destinationWriteStartTime" : 1712842650012,
    "meanSecondsBeforeSourceStateMessageEmitted" : 0,
    "maxSecondsBeforeSourceStateMessageEmitted" : 0,
    "meanSecondsBetweenStateMessageEmittedandCommitted" : 0,
    "recordsEmitted" : 0,
    "replicationEndTime" : 1712842667361,
    "replicationStartTime" : 1712842649996,
    "sourceReadEndTime" : 0,
    "sourceReadStartTime" : 1712842650015,
    "sourceStateMessagesEmitted" : 0
  },
  "streamStats" : [ ],
  "performanceMetrics" : {
    "processFromSource" : {
      "elapsedTimeInNanos" : 75674150,
      "executionCount" : 3,
      "avgExecTimeInNanos" : 2.5224716666666668E7
    },
    "readFromSource" : {
      "elapsedTimeInNanos" : 9081713300,
      "executionCount" : 18077,
      "avgExecTimeInNanos" : 502390.5128063285
    },
    "processFromDest" : {
      "elapsedTimeInNanos" : 0,
      "executionCount" : 0,
      "avgExecTimeInNanos" : "NaN"
    },
    "writeToDest" : {
      "elapsedTimeInNanos" : 0,
      "executionCount" : 0,
      "avgExecTimeInNanos" : "NaN"
    },
    "readFromDest" : {
      "elapsedTimeInNanos" : 10360120293,
      "executionCount" : 54113,
      "avgExecTimeInNanos" : 191453.4454382496
    }
  }
}
2024-04-11 13:37:47 replication-orchestrator > failures: [ {
  "failureOrigin" : "source",
  "failureType" : "config_error",
  "internalMessage" : "io.airbyte.commons.exceptions.ConfigErrorException: Saved offset is not valid. Please reset the connection, and then increase oplog retention and/or increase sync frequency to prevent his from happening in the future. See https://docs.airbyte.com/integrations/sources/mongodb-v2#mongodb-oplog-and-change-streams for more details",
  "externalMessage" : "Saved offset is not valid. Please reset the connection, and then increase oplog retention and/or increase sync frequency to prevent his from happening in the future. See https://docs.airbyte.com/integrations/sources/mongodb-v2#mongodb-oplog-and-change-streams for more details",
  "metadata" : {
    "attemptNumber" : 0,
    "jobId" : 136,
    "from_trace_message" : true,
    "connector_command" : "read"
  },
  "stacktrace" : "io.airbyte.commons.exceptions.ConfigErrorException: Saved offset is not valid. Please reset the connection, and then increase oplog retention and/or increase sync frequency to prevent his from happening in the future. See https://docs.airbyte.com/integrations/sources/mongodb-v2#mongodb-oplog-and-change-streams for more details\n\tat io.airbyte.integrations.source.mongodb.cdc.MongoDbCdcInitializer.createCdcIterators(MongoDbCdcInitializer.java:124)\n\tat io.airbyte.integrations.source.mongodb.MongoDbSource.read(MongoDbSource.java:155)\n\tat io.airbyte.cdk.integrations.base.IntegrationRunner.readSerial(IntegrationRunner.kt:318)\n\tat io.airbyte.cdk.integrations.base.IntegrationRunner.runInternal(IntegrationRunner.kt:187)\n\tat io.airbyte.cdk.integrations.base.IntegrationRunner.run(IntegrationRunner.kt:115)\n\tat io.airbyte.integrations.source.mongodb.MongoDbSource.main(MongoDbSource.java:52)\n",
  "timestamp" : 1712842664857
}, {
  "failureOrigin" : "source",
  "failureType" : "system_error",
  "internalMessage" : "io.airbyte.commons.exceptions.ConfigErrorException: Saved offset is not valid. Please reset the connection, and then increase oplog retention and/or increase sync frequency to prevent his from happening in the future. See https://docs.airbyte.com/integrations/sources/mongodb-v2#mongodb-oplog-and-change-streams for more details",
  "externalMessage" : "Something went wrong in the connector. See the logs for more details.",
  "metadata" : {
    "attemptNumber" : 0,
    "jobId" : 136,
    "from_trace_message" : true,
    "connector_command" : "read"
  },
  "stacktrace" : "io.airbyte.commons.exceptions.ConfigErrorException: Saved offset is not valid. Please reset the connection, and then increase oplog retention and/or increase sync frequency to prevent his from happening in the future. See https://docs.airbyte.com/integrations/sources/mongodb-v2#mongodb-oplog-and-change-streams for more details\n\tat io.airbyte.integrations.source.mongodb.cdc.MongoDbCdcInitializer.createCdcIterators(MongoDbCdcInitializer.java:124)\n\tat io.airbyte.integrations.source.mongodb.MongoDbSource.read(MongoDbSource.java:155)\n\tat io.airbyte.cdk.integrations.base.IntegrationRunner.readSerial(IntegrationRunner.kt:318)\n\tat io.airbyte.cdk.integrations.base.IntegrationRunner.runInternal(IntegrationRunner.kt:187)\n\tat io.airbyte.cdk.integrations.base.IntegrationRunner.run(IntegrationRunner.kt:115)\n\tat io.airbyte.integrations.source.mongodb.MongoDbSource.main(MongoDbSource.java:52)\n",
  "timestamp" : 1712842664888
}, {
  "failureOrigin" : "source",
  "internalMessage" : "Source process exited with non-zero exit code 1",
  "externalMessage" : "Something went wrong within the source connector",
  "metadata" : {
    "attemptNumber" : 0,
    "jobId" : 136,
    "connector_command" : "read"
  },
  "stacktrace" : "io.airbyte.workers.internal.exception.SourceException: Source process exited with non-zero exit code 1\n\tat io.airbyte.workers.general.BufferedReplicationWorker.readFromSource(BufferedReplicationWorker.java:362)\n\tat io.airbyte.workers.general.BufferedReplicationWorker.lambda$runAsyncWithHeartbeatCheck$3(BufferedReplicationWorker.java:234)\n\tat java.base/java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1804)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)\n\tat java.base/java.lang.Thread.run(Thread.java:1583)\n",
  "timestamp" : 1712842666444
} ]
2024-04-11 13:37:47 replication-orchestrator > Returning output...
2024-04-11 13:37:47 replication-orchestrator > 
2024-04-11 13:37:47 replication-orchestrator > ----- END REPLICATION -----

Contribute

[ ] Yes, I want to contribute