feat(source-slack): add configurable toggle to skip replies for zero-reply messages (#75905)

devin-ai-integration[bot] · bot_apk · sophiecuiy · web-flow · commit 7db08ee64643 · 2026-04-02T13:29:10.000-07:00
Co-authored-by: Devin AI &lt;158243242+devin-ai-integration[bot]@users.noreply.github.com&gt;
Co-authored-by: bot_apk &lt;apk@cognition.ai&gt;
Co-authored-by: Sophie Cui &lt;sophie.cui@airbyte.io&gt;
Co-authored-by: Claude Opus 4.6 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/airbyte-integrations/connectors/source-slack/components.py b/airbyte-integrations/connectors/source-slack/components.py
@@ -18,7 +18,6 @@
     InterpolatedString,
 )
 from airbyte_cdk.sources.declarative.migrations.state_migration import StateMigration
-from airbyte_cdk.sources.declarative.partition_routers import SinglePartitionRouter, SubstreamPartitionRouter
 from airbyte_cdk.sources.declarative.requesters import HttpRequester
 from airbyte_cdk.sources.declarative.requesters.request_options.interpolated_request_options_provider import (
     InterpolatedRequestOptionsProvider,
diff --git a/airbyte-integrations/connectors/source-slack/manifest.yaml b/airbyte-integrations/connectors/source-slack/manifest.yaml
@@ -284,6 +284,16 @@ definitions:
       schema:
         $ref: "#/schemas/channel_messages"
 
+  channel_messages_with_replies_stream:
+    $ref: "#/definitions/channel_messages_stream"
+    retriever:
+      $ref: "#/definitions/channel_messages_stream/retriever"
+      record_selector:
+        $ref: "#/definitions/channel_messages_stream/retriever/record_selector"
+        record_filter:
+          type: RecordFilter
+          condition: "{{ (record.reply_count or 0) > 0 if config.get('threads_ignore_no_replies', false) else true }}"
+
   threads_stream:
     $ref: "#/definitions/stream_base"
     $parameters:
@@ -328,7 +338,7 @@ definitions:
         parent_stream_configs:
           - type: ParentStreamConfig
             stream:
-              $ref: "#/definitions/channel_messages_stream"
+              $ref: "#/definitions/channel_messages_with_replies_stream"
             # One complication is that threads can be updated at Any time in the future. Therefore, if we wanted to comprehensively sync data
             # i.e: get every single response in a thread, we'd have to read every message in the slack instance every time we ran a sync,
             # because otherwise there is no way to guarantee that a thread deep in the past didn't receive a new message.
@@ -432,6 +442,11 @@ spec:
         examples:
           - channel_one
           - channel_two
+      threads_ignore_no_replies:
+        type: boolean
+        default: false
+        title: Ignore messages with no replies in threads stream
+        description: When enabled, the threads stream will skip messages that have no replies (reply_count is 0, null, or absent), reducing the number of API calls. Disabled by default to make Threads stream contain unthreaded messages in its records.
       credentials:
         title: Authentication mechanism
         description: Choose how to authenticate into Slack
diff --git a/airbyte-integrations/connectors/source-slack/metadata.yaml b/airbyte-integrations/connectors/source-slack/metadata.yaml
@@ -10,7 +10,7 @@ data:
   connectorSubtype: api
   connectorType: source
   definitionId: c2281cee-86f9-4a86-bb48-d23286b4c7bd
-  dockerImageTag: 3.1.14
+  dockerImageTag: 3.1.15
   dockerRepository: airbyte/source-slack
   documentationUrl: https://docs.airbyte.com/integrations/sources/slack
   externalDocumentationUrls:
diff --git a/airbyte-integrations/connectors/source-slack/unit_tests/test_streams.py b/airbyte-integrations/connectors/source-slack/unit_tests/test_streams.py
@@ -27,7 +27,7 @@ def authenticator(token_config):
         (
             "2020-01-01T00:00:00Z",
             "2020-01-02T00:00:00Z",
-            [{"ts": 1577866844}, {"ts": 1577877406}],
+            [{"ts": 1577866844, "reply_count": 1}, {"ts": 1577877406, "reply_count": 1}],
             {},
             [
                 {
@@ -95,12 +95,12 @@ def test_get_updated_state(requests_mock, authenticator, token_config, current_s
     requests_mock.register_uri(
         "GET",
         "https://slack.com/api/conversations.history?limit=1000&channel=airbyte-for-beginners",
-        [{"json": {"messages": [{"ts": 1507866847}]}}, {"json": {"messages": []}}],
+        [{"json": {"messages": [{"ts": 1507866847, "reply_count": 1}]}}, {"json": {"messages": []}}],
     )
     requests_mock.register_uri(
         "GET",
         "https://slack.com/api/conversations.history?limit=1000&channel=good-reads",
-        [{"json": {"messages": [{"ts": 1507866847}]}}, {"json": {"messages": []}}],
+        [{"json": {"messages": [{"ts": 1507866847, "reply_count": 1}]}}, {"json": {"messages": []}}],
     )
     requests_mock.register_uri(
         "GET",
@@ -142,12 +142,12 @@ def test_threads_parse_response(requests_mock, authenticator, token_config):
     requests_mock.register_uri(
         "GET",
         "https://slack.com/api/conversations.history?limit=1000&channel=airbyte-for-beginners",
-        [{"json": {"messages": [{"ts": 1507866847}]}}, {"json": {"messages": []}}],
+        [{"json": {"messages": [{"ts": 1507866847, "reply_count": 1}]}}, {"json": {"messages": []}}],
     )
     requests_mock.register_uri(
         "GET",
         "https://slack.com/api/conversations.history?limit=1000&channel=good-reads",
-        [{"json": {"messages": [{"ts": 1507866847}]}}, {"json": {"messages": []}}],
+        [{"json": {"messages": [{"ts": 1507866847, "reply_count": 1}]}}, {"json": {"messages": []}}],
     )
 
     requests_mock.register_uri(
@@ -237,12 +237,12 @@ def test_backoff(requests_mock, token_config, authenticator, headers, expected_r
     requests_mock.register_uri(
         "GET",
         "https://slack.com/api/conversations.history?limit=1000&channel=airbyte-for-beginners",
-        [{"json": {"messages": [{"ts": 1507866847}]}}, {"json": {"messages": []}}],
+        [{"json": {"messages": [{"ts": 1507866847, "reply_count": 1}]}}, {"json": {"messages": []}}],
     )
     requests_mock.register_uri(
         "GET",
         "https://slack.com/api/conversations.history?limit=1000&channel=good-reads",
-        [{"json": {"messages": [{"ts": 1507866847}]}}, {"json": {"messages": []}}],
+        [{"json": {"messages": [{"ts": 1507866847, "reply_count": 1}]}}, {"json": {"messages": []}}],
     )
 
     catalog = ConfiguredAirbyteCatalogSerializer.load(
@@ -263,6 +263,181 @@ def test_backoff(requests_mock, token_config, authenticator, headers, expected_r
     assert len(output.records) == expected_result
 
 
+def test_threads_stream_skips_messages_without_replies_when_enabled(requests_mock, token_config):
+    """
+    Verify that when threads_ignore_no_replies=True, the threads stream only creates
+    partitions for parent messages with reply_count > 0.
+    Messages with reply_count=0 or missing reply_count should be filtered out.
+    """
+    token_config["channel_filter"] = []
+    token_config["threads_ignore_no_replies"] = True
+
+    # Channel 1: one message with replies, one with reply_count=0, one with reply_count=None
+    requests_mock.register_uri(
+        "GET",
+        "https://slack.com/api/conversations.history?limit=1000&channel=airbyte-for-beginners",
+        [
+            {
+                "json": {
+                    "messages": [
+                        {"ts": 1577866844, "reply_count": 3},
+                        {"ts": 1577877406, "reply_count": 0},
+                        {"ts": 1577888888, "reply_count": None},
+                    ]
+                }
+            },
+            {"json": {"messages": []}},
+        ],
+    )
+    # Channel 2: one message with missing reply_count key entirely (should be filtered)
+    requests_mock.register_uri(
+        "GET",
+        "https://slack.com/api/conversations.history?limit=1000&channel=good-reads",
+        [
+            {"json": {"messages": [{"ts": 1577866844}]}},
+            {"json": {"messages": []}},
+        ],
+    )
+
+    stream = get_stream_by_name("threads", token_config)
+    slices = list(map(lambda partition: partition.to_slice(), stream.generate_partitions()))
+
+    # Only the message with reply_count=3 should produce a partition
+    # reply_count=0, reply_count=None, and missing reply_count should all be filtered out
+    assert len(slices) == 1
+    assert slices[0]["float_ts"] == 1577866844
+    assert slices[0]["parent_slice"]["channel"] == "airbyte-for-beginners"
+
+
+def test_threads_stream_includes_all_messages_by_default(requests_mock, token_config):
+    """
+    Verify that when threads_ignore_no_replies is not set (default=False),
+    all messages are passed through as partitions, preserving current behavior.
+    This includes messages with reply_count=None (null from API).
+    """
+    token_config["channel_filter"] = []
+    # Do NOT set threads_ignore_no_replies — should default to False
+
+    # Channel 1: messages with various reply_count values including None
+    requests_mock.register_uri(
+        "GET",
+        "https://slack.com/api/conversations.history?limit=1000&channel=airbyte-for-beginners",
+        [
+            {
+                "json": {
+                    "messages": [
+                        {"ts": 1577866844, "reply_count": 3},
+                        {"ts": 1577877406, "reply_count": 0},
+                        {"ts": 1577888888, "reply_count": None},
+                    ]
+                }
+            },
+            {"json": {"messages": []}},
+        ],
+    )
+    # Channel 2: one message with missing reply_count key
+    requests_mock.register_uri(
+        "GET",
+        "https://slack.com/api/conversations.history?limit=1000&channel=good-reads",
+        [
+            {"json": {"messages": [{"ts": 1577866844}]}},
+            {"json": {"messages": []}},
+        ],
+    )
+
+    stream = get_stream_by_name("threads", token_config)
+    slices = list(map(lambda partition: partition.to_slice(), stream.generate_partitions()))
+
+    # All 4 messages should produce partitions (no filtering when disabled)
+    assert len(slices) == 4
+
+
+def test_threads_stream_no_replies_api_calls_skipped_when_enabled(requests_mock, token_config):
+    """
+    End-to-end test: when threads_ignore_no_replies=True, verify that conversations.replies
+    is only called for messages with reply_count > 0 (not for reply_count=0, None, or absent).
+    Uses requests_mock.request_history to confirm API call reduction.
+    """
+    token_config["channel_filter"] = []
+    token_config["threads_ignore_no_replies"] = True
+
+    # Channel 1: one message with replies (reply_count=3), one without (reply_count=0)
+    requests_mock.register_uri(
+        "GET",
+        "https://slack.com/api/conversations.history?limit=1000&channel=airbyte-for-beginners",
+        [
+            {
+                "json": {
+                    "messages": [
+                        {"ts": "1577866844.000000", "reply_count": 3},
+                        {"ts": "1577877406.000000", "reply_count": 0},
+                    ]
+                }
+            },
+            {"json": {"messages": []}},
+        ],
+    )
+    # Channel 2: one message with reply_count=None (should be filtered)
+    requests_mock.register_uri(
+        "GET",
+        "https://slack.com/api/conversations.history?limit=1000&channel=good-reads",
+        [
+            {"json": {"messages": [{"ts": "1577866844.000000", "reply_count": None}]}},
+            {"json": {"messages": []}},
+        ],
+    )
+    # Only the message with reply_count=3 should trigger a conversations.replies call
+    requests_mock.register_uri(
+        "GET",
+        "https://slack.com/api/conversations.replies?channel=airbyte-for-beginners&limit=1000&ts=1577866844.000000",
+        json={
+            "messages": [
+                {
+                    "type": "message",
+                    "ts": "1577866844.000000",
+                    "thread_ts": "1577866844.000000",
+                    "reply_count": 3,
+                    "text": "parent",
+                },
+                {
+                    "type": "message",
+                    "ts": "1577866845.000000",
+                    "thread_ts": "1577866844.000000",
+                    "text": "reply",
+                },
+            ]
+        },
+    )
+
+    catalog = ConfiguredAirbyteCatalogSerializer.load(
+        {
+            "streams": [
+                {
+                    "stream": {
+                        "name": "threads",
+                        "json_schema": {},
+                        "supported_sync_modes": ["full_refresh", "incremental"],
+                    },
+                    "sync_mode": "incremental",
+                    "destination_sync_mode": "append",
+                }
+            ]
+        }
+    )
+    state = StateBuilder().with_stream_state("threads", {}).build()
+    source_slack = get_source(token_config, "threads", state)
+    output = read(source_slack, config=token_config, catalog=catalog, state=state)
+
+    # Verify records were returned from the one valid thread
+    assert len(output.records) == 2
+
+    # Verify conversations.replies was called exactly once (only for reply_count=3 message)
+    replies_calls = [req for req in requests_mock.request_history if "conversations.replies" in req.url]
+    assert len(replies_calls) == 1
+    assert "channel=airbyte-for-beginners" in replies_calls[0].url
+    assert "ts=1577866844.000000" in replies_calls[0].url
+
+
 def test_channels_stream_with_autojoin(token_config, requests_mock) -> None:
     """
     The test uses the `conversations_list` fixture(autouse=true) as API mocker.
diff --git a/docs/integrations/sources/slack.md b/docs/integrations/sources/slack.md
@@ -85,7 +85,10 @@ If you are using an API key to authenticate to Slack, a refresh token is not req
 <FieldAnchor field="include_private_channels">
 9. (Optional) **Include_private_channels** Toggle on to sync data from private channels. You will need to manually add the bot to private channels even if `join_channels` is toggled on.
 </FieldAnchor>
-10. Click **Set up source**. You must add the App created in Step 1 to the channels with the data that you want to sync.
+<FieldAnchor field="threads_ignore_no_replies">
+10. (Optional) **Ignore messages with no replies in threads stream** Toggle on to skip messages with no replies (`reply_count=0`) in the Threads stream. This reduces unnecessary `conversations.replies` API calls and can significantly speed up syncs for workspaces with many messages. Disabled by default to make the Threads stream contain unthreaded messages in its records.
+</FieldAnchor>
+11. Click **Set up source**. You must add the App created in Step 1 to the channels with the data that you want to sync.
 <!-- /env:cloud -->
 
 <!-- env:oss -->
@@ -101,7 +104,8 @@ If you are using an API key to authenticate to Slack, a refresh token is not req
 7. **Threads Lookback window (Days)**. This corresponds to the number of days in the past from which you want to sync data.
 8. (Optional) **Channel filter** the list of channel names (without leading '#' char) that limits the channels from which you'd like to sync. If no channels are specified, Airbyte will replicate data from all channels.
 9. (Optional) **Include_private_channels** Toggle on to sync data from private channels. You will need to manually add the bot to private channels even if `join_channels` is toggled on.
-10. Click **Set up source**. You must add the App created in Step 1 to the channels with the data that you want to sync.
+10. (Optional) **Ignore messages with no replies in threads stream** Toggle on to skip messages with no replies (`reply_count=0`) in the Threads stream. This reduces unnecessary `conversations.replies` API calls and can significantly speed up syncs for workspaces with many messages. Disabled by default to make the Threads stream contain unthreaded messages in its records.
+11. Click **Set up source**. You must add the App created in Step 1 to the channels with the data that you want to sync.
 <!-- /env:oss -->
 
 <HideInUI>
@@ -170,6 +174,12 @@ These two streams are effectively limited to **one request per minute**. Conside
 
 - Check out common troubleshooting issues for the Slack source connector on our Airbyte Forum [here](https://github.com/airbytehq/airbyte/discussions).
 
+#### Threads stream performance
+
+If your Threads stream syncs are slow, consider enabling the **Ignore messages with no replies in threads stream** (`threads_ignore_no_replies`) option. By default, the Threads stream calls the `conversations.replies` API for every message, including those with no replies. In many workspaces, the majority of messages have no replies, so these API calls are wasted and consume rate-limit budget.
+
+- **Set to `true`** when you want to optimize sync performance and only need thread replies for messages that actually have threaded conversations. This can reduce API calls by up to 89% depending on your workspace.
+- **Keep as `false` (default)** when you need the Threads stream to include records for all messages, including unthreaded ones. This preserves the current behavior where every message appears in the Threads stream output.
 
 </details>
 
@@ -183,6 +193,7 @@ These two streams are effectively limited to **one request per minute**. Conside
 
 | Version    | Date       | Pull Request                                             | Subject                                                                                                                                                                |
 |:-----------|:-----------|:---------------------------------------------------------|:-----------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 3.1.15 | 2026-03-31 | [75905](https://github.com/airbytehq/airbyte/pull/75905) | Add configurable option to skip conversations.replies API calls for messages with no replies, reducing unnecessary API usage |
 | 3.1.14 | 2026-03-27 | [75197](https://github.com/airbytehq/airbyte/pull/75197) | Add declarative OAuth with `oauth_connector_input_specification` and granular scopes |
 | 3.1.13 | 2026-03-24 | [75329](https://github.com/airbytehq/airbyte/pull/75329) | Update dependencies |
 | 3.1.12 | 2026-03-10 | [74598](https://github.com/airbytehq/airbyte/pull/74598) | Update dependencies |

Original file line number	Diff line number	Diff line change
`@@ -18,7 +18,6 @@`
`18`	`18`	`InterpolatedString,`
`19`	`19`	`)`
`20`	`20`	`from airbyte_cdk.sources.declarative.migrations.state_migration import StateMigration`
`21`		`-from airbyte_cdk.sources.declarative.partition_routers import SinglePartitionRouter, SubstreamPartitionRouter`
`22`	`21`	`from airbyte_cdk.sources.declarative.requesters import HttpRequester`
`23`	`22`	`from airbyte_cdk.sources.declarative.requesters.request_options.interpolated_request_options_provider import (`
`24`	`23`	`InterpolatedRequestOptionsProvider,`