[refactor-opportunist] Consolidate queue factory resolution in publisher pipeline constructors

🏗️ Refactor Proposal

Summary: Extract one shared queue-factory resolution path used by both pipeline constructors to reduce constructor drift and keep queue policy changes in one place.

Problem

libbeat/publisher/pipeline/pipeline.go currently has two constructor paths (New and NewForReceiver) that both resolve queue type and construct a queue factory. This logic was duplicated in both paths and is easy to let drift when queue behavior evolves.

Concrete evidence:

• libbeat/publisher/pipeline/pipeline.go:168 calls queue setup for New(...).
• libbeat/publisher/pipeline/pipeline.go:198 calls queue setup for NewForReceiver(...).
• Shared queue behavior is centralized lower in the same file at libbeat/publisher/pipeline/pipeline.go:353-370 (queueFactoryForUserConfig), so duplicated pre-processing in constructors is structural overhead.
• Recent churn signal (last 60 days): libbeat/publisher/pipeline/pipeline.go appears among top frequently changed files (6 touches from git log --since="60 days ago" --name-only).

Proposed Approach

Introduce a small shared helper for constructor-level queue namespace resolution and have both constructors call it:

• Keep existing queueFactoryForUserConfig as the single lower-level parser/factory constructor.
• Add queueFactoryForNamespace(...) for the repeated namespace->queueType normalization.
• Route both New(...) and NewForReceiver(...) through this helper.

This keeps behavior unchanged while reducing duplication and future divergence risk.

Proof of Concept

I partially implemented this refactor on one representative slice to verify viability:

Files changed:

• libbeat/publisher/pipeline/pipeline.go

Before → After:

• Before, both constructors had inline blocks equivalent to:
  • queueType := defaultQueueType
  • if b := userQueueConfig.Name(); b != "" { queueType = b }
  • queueFactoryForUserConfig(queueType, userQueueConfig.Config(), ...)
• After:
  • libbeat/publisher/pipeline/pipeline.go:168 and :198 now call queueFactoryForNamespace(userQueueConfig, ...).
  • Shared helper added at libbeat/publisher/pipeline/pipeline.go:337-347.

Verification

• go test -mod=readonly ./libbeat/publisher/pipeline -run '^TestNoBatchAssemblyOnNilTarget$' → ok github.com/elastic/beats/v7/libbeat/publisher/pipeline
• go test -mod=readonly -run '^$' ./libbeat/publisher/pipeline → ok github.com/elastic/beats/v7/libbeat/publisher/pipeline 0.006s [no tests to run]

Incremental Rollout Plan

This refactor can be completed incrementally:

1. Land constructor deduplication in libbeat/publisher/pipeline (proved in this PoC).
2. Apply the same “single conversion path” pattern to nearby queue-policy call sites where constructor-adjacent logic is repeated.
3. Add/extend focused tests around queue-type selection paths to guard against drift.

Risks and Mitigations

• Risk: Hidden behavior differences between constructor paths could be masked by consolidation.
  Mitigation: Keep helper narrow (namespace normalization only) and preserve existing lower-level factory logic unchanged.
• Risk: Future queue config changes may still split across call sites.
  Mitigation: Treat queueFactoryForNamespace as the required entry point for constructor queue resolution.

Evidence

• Code locations:
  • libbeat/publisher/pipeline/pipeline.go:168
  • libbeat/publisher/pipeline/pipeline.go:198
  • libbeat/publisher/pipeline/pipeline.go:337-347
  • libbeat/publisher/pipeline/pipeline.go:353-370
• Churn command:
  • git --no-pager log --since="60 days ago" --name-only --pretty=format: | sed '/^$/d' | sort | uniq -c | sort -nr
• Duplicate checks:
  • GitHub issue searches for "queueFactoryForUserConfig" and "publisher/pipeline" "NewForReceiver" refactor in elastic/beats returned total_count: 0.

[!NOTE]

🔒 Integrity filter blocked 42 items

The following items were blocked because they don't meet the GitHub integrity level.

• #9600 search_issues: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #50428 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #49800 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #37366 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #50586 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #50721 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #50581 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #49762 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #50589 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #50590 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #50588 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #50562 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #50315 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #50118 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #50337 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #50650 search_pull_requests: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• ... and 26 more items

To allow these resources, lower min-integrity in your GitHub frontmatter:

tools:
  github:
    min-integrity: approved  # merged | approved | unapproved | none

---

What is this? | From workflow: Refactor Opportunist

Give us feedback! React with 🚀 if perfect, 👍 if helpful, 👎 if not.

Start in libbeat/publisher/pipeline/pipeline.go by comparing New and NewForReceiver with queueFactoryForUserConfig and the proposed queueFactoryForNamespace path. Run the focused TestNoBatchAssemblyOnNilTarget test and the package compile-only command, then verify both constructors share namespace normalization without changing queue behavior. Done means the duplicate constructor logic is consolidated and focused queue-type selection coverage is added or extended.