> ## Documentation Index
> Fetch the complete documentation index at: https://docs.open-metadata.org/llms.txt
> Use this file to discover all available pages before exploring further.

# Run the NATS JetStream Connector Externally

> Use YAML to configure NATS JetStream metadata ingestion. Ingest JetStream streams as topics, fetch sample data, and read schemas from a KV bucket.

export const CodePanel = ({children, fileName = 'config.yaml', showLineNumbers = false}) => {
  const codePanelRef = useRef(null);
  const codeContentRef = useRef(null);
  const isProgrammaticScroll = useRef(false);
  const hoverTimeout = useRef(null);
  useEffect(() => {
    let tries = 0;
    const wrapLines = () => {
      const root = codeContentRef.current;
      if (!root) return;
      const pres = Array.from(root.querySelectorAll('pre'));
      if (!pres.length) {
        if (tries++ < 20) requestAnimationFrame(wrapLines);
        return;
      }
      let globalLine = 1;
      pres.forEach(pre => {
        const code = pre.querySelector('code') || pre;
        if (!code || code.dataset.wrapped === 'true') return;
        const raw = code.textContent || '';
        let lines = raw.split('\n');
        while (lines[0] === '') lines.shift();
        while (lines[lines.length - 1] === '') lines.pop();
        code.innerHTML = lines.map(line => {
          const ln = globalLine++;
          const num = showLineNumbers ? `<span class="line-number">${ln}</span>` : '';
          const safe = line.replace(/</g, '&lt;').replace(/>/g, '&gt;') || ' ';
          return `<span class="code-line" data-line="${ln}">${num}${safe}</span>`;
        }).join('');
        code.dataset.wrapped = 'true';
      });
    };
    wrapLines();
  }, [children, showLineNumbers]);
  useEffect(() => {
    const panel = codePanelRef.current;
    const content = codeContentRef.current;
    if (!panel || !content) return;
    const waitForLines = () => {
      const codeLines = content.querySelectorAll('.code-line');
      if (!codeLines.length) {
        requestAnimationFrame(waitForLines);
        return;
      }
      setupHighlighting(codeLines);
    };
    const setupHighlighting = codeLines => {
      const layout = panel.closest('.split-layout');
      const sections = layout.querySelectorAll('.content-section');
      const parseLines = str => {
        if (!str) return [];
        const out = [];
        str.split(',').forEach(p => {
          if (p.includes('-')) {
            const [s, e] = p.split('-').map(Number);
            for (let i = s; i <= e; i++) out.push(i);
          } else {
            const n = Number(p);
            if (!isNaN(n)) out.push(n);
          }
        });
        return out;
      };
      const clearHighlight = () => {
        codeLines.forEach(l => l.classList.remove('highlighted'));
      };
      const highlight = lines => {
        clearHighlight();
        lines.forEach(n => {
          const el = content.querySelector(`.code-line[data-line="${n}"]`);
          if (el) el.classList.add('highlighted');
        });
      };
      const scrollToLines = lines => {
        if (!lines.length) return;
        const first = lines[0];
        const targetLine = lines.length > 1 ? first : lines[0];
        const el = content.querySelector(`.code-line[data-line="${targetLine}"]`);
        if (!el) return;
        isProgrammaticScroll.current = true;
        const containerRect = content.getBoundingClientRect();
        const elRect = el.getBoundingClientRect();
        const offset = elRect.top - containerRect.top + content.scrollTop;
        const TOP_PADDING = 16;
        content.scrollTo({
          top: Math.max(offset - TOP_PADDING, 0),
          behavior: 'smooth'
        });
        setTimeout(() => {
          isProgrammaticScroll.current = false;
        }, 200);
      };
      const activate = (section, scroll) => {
        if (section.classList.contains('active')) return;
        sections.forEach(s => s.classList.remove('active'));
        section.classList.add('active');
        const lines = parseLines(section.dataset.lines);
        highlight(lines);
        if (scroll) scrollToLines(lines);
      };
      const observer = new IntersectionObserver(entries => {
        if (isProgrammaticScroll.current) return;
        entries.forEach(e => {
          if (e.isIntersecting) activate(e.target, false);
        });
      }, {
        threshold: 0.3,
        rootMargin: '-80px 0px -40% 0px'
      });
      sections.forEach(section => {
        observer.observe(section);
        section.addEventListener('click', () => activate(section, true));
        section.addEventListener('mouseenter', () => {
          clearTimeout(hoverTimeout.current);
          hoverTimeout.current = setTimeout(() => activate(section, true), 80);
        });
      });
      if (sections[0]) activate(sections[0], false);
    };
    waitForLines();
  }, []);
  const handleCopy = e => {
    const btn = e.currentTarget;
    const codeLines = codeContentRef.current?.querySelectorAll('.code-line');
    if (!codeLines || codeLines.length === 0) return;
    const text = Array.from(codeLines).map(line => {
      const clone = line.cloneNode(true);
      const lineNumber = clone.querySelector('.line-number');
      if (lineNumber) lineNumber.remove();
      return clone.textContent;
    }).join('\n');
    if (!text) return;
    navigator.clipboard.writeText(text).then(() => {
      btn.dataset.copied = 'true';
      setTimeout(() => btn.dataset.copied = 'false', 1500);
    });
  };
  return <div className="code-panel" ref={codePanelRef}>
      <div className="code-header">
        {fileName}
        <button className="copy-btn" aria-label="Copy full code" data-copied="false" onClick={handleCopy}>
          <svg className="icon-copy" viewBox="0 0 15 16" fill="currentColor">
            <path d="M10.113 3.124H2.205C1.463 3.124.86 3.655.86 4.31v10.005c0 .654.603 1.186 1.345 1.186h7.908c.742 0 1.345-.532 1.345-1.186V4.31c0-.655-.606-1.186-1.345-1.186Z" />
            <path d="M13.138.5H5.229c-.742 0-1.344.531-1.344 1.186 0 .23.209.414.47.414s.47-.184.47-.414c0-.197.182-.357.404-.357h7.909c.223 0 .404.16.404.357V11.69c0 .196-.181.356-.404.356-.262 0-.47.184-.47.415 0 .23.208.415.47.415.742 0 1.344-.532 1.344-1.186V1.686C14.482 1.03 13.88.5 13.138.5Z" />
          </svg>

          <svg className="icon-check" viewBox="0 0 20 20" fill="currentColor">
            <path fillRule="evenodd" d="M16.707 5.293a1 1 0 010 1.414l-7.25 7.25a1 1 0 01-1.414 0l-3.25-3.25a1 1 0 011.414-1.414l2.543 2.543 6.543-6.543a1 1 0 011.414 0z" clipRule="evenodd" />
          </svg>
        </button>
      </div>

      <div className="code-content" ref={codeContentRef}>
        {children}
      </div>
    </div>;
};

export const ContentSection = ({id, title, lines, children}) => <div className="content-section" data-content-id={id} data-lines={lines}>
    {title && <h4>{title}</h4>}
    {children}
  </div>;

export const ContentPanel = ({children}) => <div className="content-panel">{children}</div>;

export const CodePreview = ({children}) => {
  const [instanceId] = useState(() => `preview-${Math.random().toString(36).slice(2)}`);
  useEffect(() => {
    const nav = document.querySelector('nav') || document.querySelector('header') || document.querySelector('[class*="nav"]');
    if (nav) {
      document.documentElement.style.setProperty('--navbar-height', `${nav.offsetHeight}px`);
    }
  }, []);
  return <div className="split-layout" data-preview-id={instanceId}>
      {children}
    </div>;
};

export const ConnectorDetailsHeader = ({name, icon, stage, availableFeatures, unavailableFeatures = [], availableFeaturesCollate = []}) => {
  const showSubHeading = availableFeatures?.length > 0 || unavailableFeatures?.length > 0 || availableFeaturesCollate?.length > 0;
  const totalAvailableFeatures = [...availableFeatures || [], ...availableFeaturesCollate || []];
  return <div className="container">
      <div className="Heading">
        <div className="flex items-center gap-3">
          {icon && <div className="IconContainer">
              <img src={icon} alt={name} noZoom className="ConnectorIcon" />
            </div>}
          <h1 className="ConnectorName">{name}</h1>
          <span className={`StageBadge ${stage === 'PROD' ? 'prod' : 'beta'}`}>
            {stage}
          </span>
        </div>
      </div>
      {showSubHeading && <div className="SubHeading">
          <div className="FeaturesHeading">Feature List</div>
          <div className="FeaturesList">
            {totalAvailableFeatures.map(feature => <div className="FeatureTag AvailableFeature" key={feature}>
                ✓ {feature}
              </div>)}
            {unavailableFeatures.map(feature => <div className="FeatureTag UnavailableFeature" key={feature}>
                ✕ {feature}
              </div>)}
          </div>
        </div>}
    </div>;
};

<ConnectorDetailsHeader icon="/public/images/connectors/nats.png" name="NATS JetStream" stage="BETA" availableFeatures={["Topics", "Sample Data"]} unavailableFeatures={[]} />

This section covers how to use the NATS JetStream connector.
Configure and run the NATS JetStream metadata workflow externally:

* [Requirements](#requirements)
* [Metadata Ingestion](#metadata-ingestion)

## How to Run the Connector Externally

To run the Ingestion via the UI you'll need to use the OpenMetadata Ingestion Container, which comes shipped with
custom Airflow plugins to handle the workflow deployment.

If, instead, you want to manage your workflows externally on your preferred orchestrator, you can check
the following docs to run the Ingestion Framework **anywhere**.

<Columns cols={2}>
  <Card title="External Schedulers" href="/v2.1.x-SNAPSHOT/deployment/ingestion">
    Get more information about running the Ingestion Framework Externally
  </Card>
</Columns>

## Requirements

The connector communicates with the NATS JetStream API, so JetStream must be enabled on your NATS server. The credentials you supply must be allowed to publish to and receive replies from the `$JS.API.>` subjects. Schema ingestion additionally requires access to the key-value (KV) stream subjects that back the configured Schema KV bucket.

### Python Requirements

Use a Python version supported by the `openmetadata-ingestion` package that matches your OpenMetadata server. To find the supported range for your release, check the `Requires-Python` metadata of that release’s ingestion package.

To run the NATS JetStream ingestion, install:

```bash theme={null}
pip3 install "openmetadata-ingestion[nats]"
```

## Metadata Ingestion

All connectors are defined as JSON Schemas.
See the [NATS connection JSON Schema](https://github.com/open-metadata/OpenMetadata/blob/main/openmetadata-spec/src/main/resources/json/schema/entity/services/connections/messaging/natsConnection.json) for the structure of a NATS connection.
To create and run a metadata ingestion workflow, follow these steps to build a YAML configuration that connects to the source, processes the entities if needed, and reaches the OpenMetadata server.
The workflow is modeled around the following
[JSON Schema](https://github.com/open-metadata/OpenMetadata/blob/main/openmetadata-spec/src/main/resources/json/schema/metadataIngestion/workflow.json)

### 1. Define the YAML Config

This is a sample config for NATS:

<CodePreview>
  <ContentPanel>
    <ContentSection id={1} title="Source Configuration" lines="1-6">
      Configure the source type and service name. `type` is `nats` and the service connection `type` is `Nats`.
    </ContentSection>

    <ContentSection id={2} title="natsServers" lines="7">
      **natsServers**: NATS server URLs as comma-separated values. Each entry uses the `nats://host:port` form (the default NATS port is `4222`).
      Example: `nats://host1:4222,nats://host2:4222`. This is the only required field.
    </ContentSection>

    <ContentSection id={3} title="authType" lines="8-17">
      **authType**: The authentication method. Omit this block entirely for anonymous access. Otherwise provide exactly one of:

      * **Username and Password**: `username` and `password`.
      * **Token**: A single `token`.
      * **NKey Seed**: An `nkeySeed` (the account seed, which begins with `S`).
    </ContentSection>

    <ContentSection id={4} title="schemaKvBucket" lines="18">
      **schemaKvBucket**: Name of the JetStream KV bucket where schemas are stored. Keys must match stream (topic) names and values should be Avro JSON, Protobuf (`.proto`), or JSON Schema text. Leave unset to skip schema ingestion.
    </ContentSection>

    <ContentSection id={5} title="additionalConfig" lines="19">
      **additionalConfig**: Additional NATS client configuration options passed through to the client. See the [NATS Python client documentation](https://nats-io.github.io/nats.py/). The following connection-managed keys are reserved and can't be set here: `servers`, `user`, `password`, `token`, `nkeys_seed`, `nkeys_seed_str`, `tls`, `user_credentials`, `signature_cb`, and `user_jwt_cb`.
    </ContentSection>

    <ContentSection id={6} title="tlsConfig" lines="20-24">
      **tlsConfig**: Transport Layer Security (TLS) and Secure Sockets Layer (SSL) configuration for secure connections. Provide `caCertificate` to validate the server. For mutual TLS, also provide `sslCertificate` (client certificate) and `sslKey` (its private key). Both must be set together.
    </ContentSection>

    <ContentSection id={7} title="topicFilterPattern" lines="25-27">
      **topicFilterPattern**: Regex that limits ingestion to streams matching the pattern. Use `includes` and/or `excludes`.
    </ContentSection>

    <ContentSection id={8} title="supportsMetadataExtraction" lines="28">
      **supportsMetadataExtraction**: Set `supportsMetadataExtraction` to `true` or `false`.
    </ContentSection>

    <ContentSection id={9} title="Source Config" lines="29-40">
      #### Source Configuration - Source Config

      The `sourceConfig` is defined [here](https://github.com/open-metadata/OpenMetadata/blob/main/openmetadata-spec/src/main/resources/json/schema/metadataIngestion/messagingServiceMetadataPipeline.json):

      * **generateSampleData**: Option to turn on or off generating sample data during metadata extraction. It supports boolean values either `true` or `false`.
      * **topicFilterPattern**: Note that the `topicFilterPattern` supports regex as include or exclude.
      * **markDeletedTopics**: Optional configuration to soft delete topics in OpenMetadata if the source topics are deleted. Also, if the topic is deleted, all the associated entities like sample data, lineage, etc., with that topic will be deleted. It supports boolean values either `true` or `false`.
      * **overrideMetadata**: Set the **Override Metadata** toggle to control whether to override the existing metadata in the OpenMetadata server with the metadata fetched from the source. If the toggle is set to true, the metadata fetched from the source will override the existing metadata in the OpenMetadata server. If the toggle is set to false, the metadata fetched from the source will not override the existing metadata in the OpenMetadata server. This is applicable for fields like description, tags, owner, and displayName. It supports boolean values either `true` or `false`.
    </ContentSection>

    <ContentSection id={10} title="Sink Configuration" lines="41-43">
      To send the metadata to OpenMetadata, it needs to be specified as `type: metadata-rest`.
    </ContentSection>

    <ContentSection id={11} title="Workflow Configuration" lines="44-60">
      The main property here is `openMetadataServerConfig`, where you can define the host and security provider of your OpenMetadata installation.

      * **loggerLevel**: Specify the logger level depending on your needs. If you are troubleshooting an ingestion, use `DEBUG` for more detailed traces.
      * **JWT token**: JWT tokens allow clients to authenticate against the OpenMetadata server. See [Enable JWT Tokens](/deployment/security/enable-jwt-tokens) and [JWT Troubleshooting](/deployment/security/jwt-troubleshooting) for more information.
      * **storeServiceConnection**: If set to `true` (default), sensitive information is stored encrypted with the Fernet Key or externally if you have configured a [Secrets Manager](/deployment/secrets-manager). If set to `false`, the service is created, but the service connection information is only used by the Ingestion Framework at runtime and is not sent to the OpenMetadata server.
      * **SSL configuration**: If you have added SSL to the [OpenMetadata server](/deployment/security/enable-ssl), configure the certificates for ingestion. Set `verifySSL` to `ignore`, or set it to `validate` and provide `sslConfig.caCertificate` with a local path to the server certificate. See [SSL Troubleshooting](/deployment/security/enable-ssl/ssl-troubleshooting) for more information.
      * **ingestionPipelineFQN**: Fully qualified name of the ingestion pipeline, used to identify the current ingestion pipeline.
    </ContentSection>
  </ContentPanel>

  <CodePanel fileName="nats_config.yaml">
    ```yaml theme={null}
    source:
      type: nats
      serviceName: local_nats
      serviceConnection:
        config:
          type: Nats
          natsServers: nats://localhost:4222  # comma-separated for multiple servers
          # Basic auth (username and password). Omit authType entirely for anonymous access.
          authType:
            username: <username>
            password: <password>
          # Token auth:
          # authType:
          #   token: <token>
          # NKey auth:
          # authType:
          #   nkeySeed: <nkey_seed>
          # schemaKvBucket: <kv_bucket_name>  # uncomment to enable schema ingestion
          additionalConfig: {}
          # TLS / mTLS configuration:
          # tlsConfig:
          #   caCertificate: "-----BEGIN CERTIFICATE-----\n...\n-----END CERTIFICATE-----"
          #   sslCertificate: "-----BEGIN CERTIFICATE-----\n...\n-----END CERTIFICATE-----"
          #   sslKey: "-----BEGIN RSA PRIVATE KEY-----\n...\n-----END RSA PRIVATE KEY-----"
          # topicFilterPattern:
          #   includes:
          #     - "^orders.*"
          # supportsMetadataExtraction: true
    ```

    ```yaml theme={null}
      sourceConfig:
        config:
          type: MessagingMetadata
          topicFilterPattern:
            excludes:
              - _confluent.*
          # includes:
          #   - topic1
          # generateSampleData: true
          # generateSampleData: false # true
          # markDeletedTopics: true # false
          # overrideMetadata:  false # true

    ```

    ```yaml theme={null}
    sink:
      type: metadata-rest
      config: {}
    ```

    ```yaml theme={null}
    workflowConfig:
      loggerLevel: INFO  # DEBUG, INFO, WARNING or ERROR
      openMetadataServerConfig:
        hostPort: "http://localhost:8585/api"
        authProvider: openmetadata
        securityConfig:
          jwtToken: "{bot_jwt_token}"
        ## Store the service Connection information
        storeServiceConnection: true  # false
        ## Secrets Manager Configuration
        # secretsManagerProvider: aws, azure or noop
        # secretsManagerLoader: airflow or env
        ## If SSL, fill the following
        # verifySSL: validate  # or ignore
        # sslConfig:
        #   caCertificate: /local/path/to/certificate
    # ingestionPipelineFQN: <service name>.<ingestion name> ## e.g., "my_redshift.metadata"
    ```
  </CodePanel>
</CodePreview>

### 2. Run with the CLI

First, we will need to save the YAML file. Afterward, and with all requirements installed, we can run:

```bash theme={null}
metadata ingest -c <path-to-yaml>
```

Note that from connector to connector, this recipe will always be the same. By updating the YAML configuration,
you will be able to extract metadata from different sources.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.