Class DatasourceManager

java.lang.Object
ai.pipestream.module.pipelineprobe.pipelinecrawl.DatasourceManager
@ApplicationScoped public class DatasourceManager extends Object
Creates a datasource for a RunPipelineCrawl run and returns its id + api key directly. Plain blocking call on a VT — no observers, no futures.

Connector selection is deterministic per source: each PipelineSource maps to a single connector_type string ("jdbc", "s3"). We look up the connector by type via DataSourceAdminServiceGrpc.DataSourceAdminServiceBlockingStub.listConnectorTypes(ListConnectorTypesRequest) and use its real connector_id. No "default" connector, no fallback id, no hardcoded UUIDs — if the platform doesn't have a connector of the requested type seeded, we fail loudly.

  • Constructor Details

    • DatasourceManager

      public DatasourceManager()
      Creates the datasource manager. The gRPC stubs are injected by the container; instances are managed by CDI.

  • Method Details

    • createDatasource

      public DatasourceManager.DatasourceCredentials createDatasource(String accountId, ai.pipestream.testing.harness.v1.PipelineSource source)
      Get-or-create a datasource under the given account, bound to the connector type that matches the requested PipelineSource.

      The datasource id is deterministic (hash(account_id + connector_id)), so reruns under a stable account converge on the same datasource — and therefore the same graph (graphId is account-derived) and the same idempotent intake-node binding. We reuse an existing datasource when present (its api key is returned by the admin service) instead of churning a new one each run. A fresh/random account naturally has none and gets a freshly-created datasource.

      Parameters:
      accountId - the account the datasource is created/reused under
      source - the requested pipeline source, mapped to a connector type
      Returns:
      the datasource's id + api key (reused or freshly created)
      Throws:
      IllegalStateException - if the connector type isn't registered or the connector reports success=false

    • ensureConnectorType

      public void ensureConnectorType(String connectorType, String displayName)
      Idempotently register a connector TYPE — for harnesses that act as their own connector (the engine-perf test registers testing-sidecar). ALREADY-EXISTS (by any signal) is success.
      Parameters:
      connectorType - the connector_type string to register
      displayName - the human-readable name for the connector type
      Throws:
      IllegalStateException - if registration fails for a reason other than the type already existing

    • rotateApiKey

      public String rotateApiKey(String datasourceId)
      Rotate the datasource's API key and return the NEW key. Doubles as a run lock for single-writer test harnesses: any still-open stream from a previous run fails its next handshake on the old key.
      Parameters:
      datasourceId - the datasource whose api key is rotated (old key invalidated immediately)
      Returns:
      the new api key
      Throws:
      IllegalStateException - if the rotate call reports success=false

    • createDatasourceForType

      public DatasourceManager.DatasourceCredentials createDatasourceForType(String accountId, String connectorType)
      Get-or-create the (account, connector-type) datasource — the PipelineSource-agnostic core of createDatasource(String, PipelineSource).
      Parameters:
      accountId - the account the datasource is created/reused under
      connectorType - the connector_type whose registered connector the datasource binds to
      Returns:
      the datasource's id + api key (reused or freshly created)
      Throws:
      IllegalStateException - if no connector of the type is registered or creation reports success=false