Source Hubspot: deals stream doesn`t sync associations when sync mode is Incremental + Dedupe

[coadan]

Environment

• Airbyte version: 0.39.29-alpha
• Deployment: Kubernetes
• Source Connector and version: HubSpot 0.1.72
• Destination Connector and version: Postgres 0.4.28
• Normalization: No normalization used, only emitting JSON documents to the destination.

Current Behavior

The current behavior is that when Incremental + Dedupe sync mode is enabled for the Deals stream in HubSpot, then the associations for the deals are missing in the final JSON document.

Currently:

{
  "id": "4830871788",
  "archived": false,
  "createdAt": "2022-06-13T09:01:25.327Z",
  "updatedAt": "2022-06-13T09:01:25.730Z",
  "properties": { }
}

Expected Behavior

The deal JSON document should contain the associated companies and contacts. This works when Full Refresh - Overwrite is selected as sync mode, and emitts the expected result as follows:

{
  "id": "4768283850",
  "archived": false,
  "contacts": [
    "176157"
  ],
  "companies": [
    "5686664638",
    "5686664638"
  ],
  "createdAt": "2022-06-03T16:04:46.510Z",
  "updatedAt": "2022-06-03T16:04:47.147Z",
  "properties": { }
}

Steps to Reproduce

Create a connection between HubSpot <-> Postgres with sync-mode Incremental + Dedupe on Deals stream and JSON output
Check if objects contain contacts and companies

[marcosmarxm]

Thanks for reporting this @coadan.

I was able to reproduce issue using integration account. Left using full refresh, right using incremental + dedup

[faacsousa]

This is happening because incremental mode uses the CRM Api | Deals | Search endpoint which doesn't have a why to fetch associations along with the response.

On full refresh the endpoint CRM Api | Deals | List is used and have the possibility to fetch associations.

I found this thread on the hubspot community forum that suggests to use this endpoint - CRM API | Associations. Specifically the 'Read a batch of associations' endpoint to solve this issue.

Not sure how to integrate with the current connector code though

[mranawake]

Commenting to subscribe for updates and report that I've experienced the same behavior on AB cloud too.

[marcosmarxm]

@davydov-d #14999 closes this issue too?

[marcosmarxm]

@mranawake can you send a message in Intercom reporting the issue? @erica-airbyte fyi

[davydov-d]

@davydov-d #14999 closes this issue too?

yes, it should

[vladimir-remar]

@davydov-d #14999 closes this issue too?

In this particulary issue it does not, or at least those streams wich are using the CRMSearchStream because it is related with wich endpoint will use as @faacsousa commented before, there is no way to retrieve the Associations when the search endpoint are used.

airbyte/airbyte-integrations/connectors/source-hubspot/source_hubspot/streams.py

Line 761 in 3fc6091

return f"/crm/v3/objects/{self.entity}/search" if self.state else f"/crm/v3/objects/{self.entity}"

[mranawake]

@marcosmarxm already messaged @erica-airbyte on intercom. she is keeping me in the loop.

[davydov-d]

well at least it wont use search endpoint unconditionally for incremental syncs after the revert is done

airbyte/airbyte-integrations/connectors/source-hubspot/source_hubspot/streams.py

Line 763 in ae0cf4c

return f"/crm/v3/objects/{self.entity}/search" if self.is_incremental_sync else f"/crm/v3/objects/{self.entity}"

[davydov-d]

well at least it wont use search endpoint unconditionally for incremental syncs after the revert is done

airbyte/airbyte-integrations/connectors/source-hubspot/source_hubspot/streams.py

Line 763 in ae0cf4c

return f"/crm/v3/objects/{self.entity}/search" if self.is_incremental_sync else f"/crm/v3/objects/{self.entity}"

but yes, that's not enough - just checked it. Hoped rolling back my PR would solve the issue but luckily the original PR was not the cause 😌

[artem1205]

Hi @coadan,
I've rechecked this issue, and cannot reproduce it. I've successfully received companies in last sync using Incremental+ append_dedup mode:

[artem1205]

Closing this issue due to inactivity, feel free to reopen if the problem still exists