const fs = require('fs')
const path = require('path')
const lodash = require('lodash')
const fg = require('fast-glob')
const cv = require('compare-versions')
const ___dirname = path.join(__dirname, '..')
const yaml = require('js-yaml')

async function getLatestVersionFiles(dir) {
  let files = await fg(['./*.md', './**/*.md'], { cwd: dir })
  files = files.filter((file) => {
    return file.split('/').length > 3
  })
  /**
   * only keep the big versions (like vX.Y) and the latest smaller version (like vX.Y.Z) that if without corresponding big version
   * for example:
   * input: ["xxx/xxx/v2.1.0/xxx.md", "xxx/xxx/v2.1.1/xxx.md", "xxx/xxx/v2.2.1/xxx.md", "xxx/xxx/v2.2.2/xxx.md"]
   * output: ["xxx/xxx/v2.1.1/xxx.md", "xxx/xxx/v2.2.2/xxx.md"]
   * the smaller versions "xxx/xxx/v2.2.0.1" will be filtered because it is lower than "v2.2.0.2"
   */
  let filesGroup = lodash.groupBy(files, (file) => {
    let [category, name, version] = file.split('/')
    if (version.split('.').length > 2) {
      const [v1, v2] = version.split('.')
      version = `${v1}.${v2}`
    }
    return `${category}-${name}-${version}`
  })
  files = Object.values(filesGroup).map((group) => {
    return group.sort((a, b) => {
      let [, , aVersion] = a.split('/')
      let [, , bVersion] = b.split('/')
      try {
        return cv.compareVersions(bVersion, aVersion)
      } catch (error) {
        return 1
      }
    })[0]
  })
  // only get the latest version
  filesGroup = lodash.groupBy(files, (file) => {
    let [category, name] = file.split('/')
    return `${category}-${name}`
  })
  const _files = Object.values(filesGroup).map((group) => {
    return group
      .sort((a, b) => {
        let [, , aVersion] = a.split('/')
        let [, , bVersion] = b.split('/')
        try {
          return cv.compareVersions(bVersion, aVersion)
        } catch (error) {
          return 1
        }
      })
      .slice(0, Math.min(1, group.length))
  })
  return lodash.flatten(_files)
}

async function addRedirect(dir) {
  dir = dir.slice(-1) === '/' ? dir : dir + '/'
  const files = await getLatestVersionFiles(dir)

  /**
   * redirects:
   *  - /category/name
   */
  for (let file of files) {
    let [category, name] = file.split('/')
    category = category.trim().toLowerCase().endsWith('s')
      ? category.substring(0, category.length - 1)
      : category

    const mdpath = dir + file
    let data = fs.readFileSync(mdpath, 'utf8')

    let yamlFile = path.join(
      path.dirname(mdpath),
      '..',
      path.basename(mdpath, '.md') + '.yaml'
    )
    let ymlData = null
    if (fs.existsSync(yamlFile)) {
      ymlData = yaml.load(fs.readFileSync(yamlFile, 'utf8'))
    }
    if (ymlData) {
      if (ymlData.icon) {
        data = data.replace(/icon: .*/, 'icon: ' + ymlData.icon)
      }
      if (ymlData.sn_available) {
        data = data.replace(/(sn_available:.*\n)/, '')
        data = data.replace(
          /(id:.*\n---)/,
          'sn_available: ' + ymlData.sn_available + '\n$1'
        )
      }
    }
    // check if `/(id:.*\n---)/g` is exist in data
    if (data.match(/(id:.*\n---)/g)) {
      data = data.replace(
        /(id:.*\n---)/g,
        'redirects:\n - /hub/' + category + '-' + name + '\n$1'
      )
    } else if (data.match(/(powered_by:.*\n---)/g)) {
      // check if `/(powered_by:.*\n---)/g` is exist in data
      data = data.replace(
        /(powered_by:.*\n---)/g,
        'redirects:\n - /hub/' + category + '-' + name + '\n$1'
      )
    }
    fs.writeFileSync(dir + file, data)
    console.log(
      dir +
        file +
        ' add redirects to ' +
        category +
        '-' +
        name +
        ' successfully'
    )
  }
}

// Example: node fix-hub-redirect.js connectors/aws-lambda-sink/2.10.0/aws-lambda-sink.md
const [dir] = process.argv.slice(2)
addRedirect(path.join(___dirname, dir))

// Usage:
// Example 1: Update a specify md file:
// node fix-hub-redirect.js connectors/aws-lambda-sink/2.10.0/aws-lambda-sink.md

// Example 2: Update files in a specify directory:
// node fix-hub-redirect.js connectors/aws-lambda-sink

// Example 3: Update all connectors docs:
// node fix-hub-redirect.js connectors

// Example 4: Update all docs:
// node fix-hub-redirect.js .


const fs = require('fs')
const path = require('path')
const ___dirname = path.join(__dirname, '..')

function travel(dir, callback) {
  if (dir.includes('node_modules')) {
    return
  }
  if (fs.statSync(dir).isFile()) {
    if (dir.endsWith('.md')) {
      callback(dir)
    }
  } else {
    fs.readdirSync(dir).forEach((file) => {
      var pathname = path.join(dir, file)
      if (fs.statSync(pathname).isDirectory()) {
        travel(pathname, callback)
      } else if (pathname.endsWith('.md')) {
        callback(pathname)
      }
    })
  }
}

function fix(mdpath) {
  let data = fs.readFileSync(mdpath, 'utf8')
  let updateData = fs.readFileSync(mdpath, 'utf8')
  let reg = /([ ]*)> \*+Note.*\n(([ ]*>\s*.*\n)*)/g
  while ((m = reg.exec(data))) {
    updateData = updateData.replace(
      m[0],
      m[1] +
        '{% callout title="Note" type="note" %}\n' +
        m[2].replace(/>[ \n]/g, '') +
        m[1] +
        '{% /callout %}\n'
    )
  }
  reg = /([ ]*)> \*+Tip.*\n(([ ]*>\s*.*\n)*)/g
  while ((m = reg.exec(data))) {
    updateData = updateData.replace(
      m[0],
      m[1] +
        '{% callout title="Tip" type="tip" %}\n' +
        m[2].replace(/>[ \n]/g, '') +
        m[1] +
        '{% /callout %}\n'
    )
  }
  fs.writeFileSync(mdpath, updateData)
  console.log(mdpath + ' fixed')
}

// Example: node fix-hub-note.js connectors/aws-lambda-sink/2.10.0/aws-lambda-sink.md
const [dir] = process.argv.slice(2)
travel(path.join(___dirname, dir), fix)

// Usage:
// Example 1: Update a specify md file:
// node fix-hub-note.js connectors/aws-lambda-sink/2.10.0/aws-lambda-sink.md

// Example 2: Update files in a specify directory:
// node fix-hub-note.js connectors/aws-lambda-sink

// Example 3: Update all connectors docs:
// node fix-hub-note.js connectors

// Example 4: Update all docs:
// node fix-hub-note.js .


const fs = require('fs')
const path = require('path')
const ___dirname = path.join(__dirname, '..')

function travel(dir, callback) {
  if (fs.statSync(dir).isFile()) {
    if (dir.endsWith('.md')) {
      callback(dir)
    }
  } else {
    fs.readdirSync(dir).forEach((file) => {
      var pathname = path.join(dir, file)
      if (fs.statSync(pathname).isDirectory()) {
        travel(pathname, callback)
      } else if (pathname.endsWith('.md')) {
        callback(pathname)
      }
    })
  }
}

function fix(mdpath) {
  let data = fs.readFileSync(mdpath, 'utf8')
  data = data
    .replace(/::: tabs/g, '{% tabs %}')
    .replace(/:::/g, '{% /tabs %}')
    .replace(/@@@[ ]+(?<tab_name>.+)/g, '{% tab label="$<tab_name>" %}')
    .replace(/@@@\s*\n/g, '{% /tab %}\n')
  fs.writeFileSync(mdpath, data)
  console.log(mdpath + ' fixed')
}

// Example: node fix-hub-tab.js connectors/aws-lambda-sink/2.10.0/aws-lambda-sink.md
const [dir] = process.argv.slice(2)
travel(path.join(___dirname, dir), fix)

// Usage:
// Example 1: Update a specify md file:
// node fix-hub-tab.js connectors/aws-lambda-sink/2.10.0/aws-lambda-sink.md

// Example 2: Update files in a specify directory:
// node fix-hub-tab.js connectors/aws-lambda-sink

// Example 3: Update all connectors docs:
// node fix-hub-tab.js connectors

// Example 4: Update all docs:
// node fix-hub-tab.js .


/*
 * NOTE! This file and the config files associated (like configCloudHosted, etc) were used for moving over the content
 * This code can eventually be removed!
 *
 */
const fs = require('fs')
const path = require('path')
const yaml = require('js-yaml')
const mkdirp = require('mkdirp')
const inFile = process.argv[2]
if (!inFile) {
  throw new Error('pass the config file')
}
const inFileFull = path.resolve(process.cwd(), inFile)
if (!fs.existsSync(inFileFull)) {
  throw new Error(`confilg ${inFile} does not exist`)
}

const configFile = JSON.parse(fs.readFileSync(inFileFull, 'utf-8'))

const sidebar = configFile.sidebar
const destFolder = configFile.destination
const pathRegexStr = configFile.pathRegex || ''
const pathReplace = configFile.pathReplace || ''
const tags = (configFile.defaultTags || '').split(',')
const idDbF = configFile.idDb

const idDbPath = path.resolve(process.cwd(), idDbF)
const sidebarPath = path.resolve(process.cwd(), sidebar)
const sourceFolder = path.dirname(sidebarPath)
const destFolderFull = path.resolve(process.cwd(), destFolder)

if (!fs.existsSync(sidebarPath)) {
  throw new Error(`input ${sidebar} does not exist`)
}
if (!fs.existsSync(destFolderFull)) {
  throw new Error(`output folder ${destFolder} does not exist`)
}
if (!fs.existsSync(idDbPath)) {
  throw new Error(`id db ${idDbF} does not exist`)
}

const sidebarJson = JSON.parse(fs.readFileSync(sidebarPath, 'utf-8'))
const idDb = JSON.parse(fs.readFileSync(idDbPath, 'utf-8'))

const FMRegex = /^---([\s\S]+?)---.*/
const PathRegex = new RegExp(pathRegexStr)
function updateFrontmatter(meta, parents, docIndex) {
  console.log(`processing ${meta.path}`)
  const docPath = path.join(sourceFolder, meta.key)
  if (!fs.existsSync(docPath)) {
    throw new Error(`document ${docPath} does not exist`)
  }

  const doc = fs.readFileSync(docPath, 'utf-8')
  const fmm = doc.match(FMRegex)
  if (!fmm && !fmm[1]) {
    throw new Error(`doc ${docPath} did not have regex`)
  }
  const fm = yaml.load(fmm[1])
  fm.oldPath = meta.path.replace(PathRegex, pathReplace)
  fm.tags = fm.tags || []
  fm.tags = fm.tags.concat(tags)
  if (meta.path.includes('connect')) {
    fm.tags.push('connect')
  }
  if (meta.path.includes('client')) {
    fm.tags.push('clients')
  }
  if (meta.path.includes('cli')) {
    fm.tags.push('tools')
  }
  if (fm.id === 'overview' && parents.length > 0) {
    console.log('FIXING generic overview ID')
    const parts = meta.key.split('/')
    parts.pop()
    const parentpart = parts.pop()
    const newid = `${parentpart}-${fm.id}`
    fm.id = newid
    idDb[fm.id] = meta.key
  } else if (idDb[fm.id] != null) {
    if (idDb[fm.id] !== meta.key) {
      const parts = meta.key.split('/').filter((r) => !r)
      parts[parts.length - 1] = fm.id
      const newid = parts.join('-')
      console.log(`CONFLICT! renaming ${meta.key} to ${newid}, ${parts}`)
      fm.id = newid
      idDb[fm.id] = meta.key
    } else {
      console.log(`REPROCESSING ${meta.key}`)
    }
  } else {
    idDb[fm.id] = meta.key
  }
  const updateDoc = doc.replace(FMRegex, `---\n${yaml.dump(fm)}\n---\n`)
  const newDir = path.join(destFolderFull, [...parents].join('/'))
  try {
    mkdirp.sync(newDir)
  } catch (ex) {}
  // replace slashes with dashes
  const metaLabel = meta.label.replace('/', '-')
  const newPath = path.join(newDir, `${100 + docIndex} - ${metaLabel}.md`)
  console.log(`wrote ${newPath}`)
  fs.writeFileSync(newPath, updateDoc)
}

function traverse(docs, paths) {
  const p = paths || []
  for (let i = 0; i < docs.length; i++) {
    const doc = docs[i]
    if (doc.children.length > 0) {
      const np = [...p, `${100 + i} - ${doc.label}`]
      traverse(doc.children, np)
    } else {
      updateFrontmatter(doc, p, i)
    }
  }
}

traverse(sidebarJson.data)

// update the db
fs.writeFileSync(idDbPath, JSON.stringify(idDb, null, 2))


const fs = require('fs')
const path = require('path')
const lodash = require('lodash')
const fg = require('fast-glob')
const cv = require('compare-versions')
const ___dirname = path.join(__dirname, '../..')
const yaml = require('js-yaml')

async function getLatestVersionFiles(dir) {
  let files = await fg(['./*.md', './**/*.md'], { cwd: dir })
  files = files.filter((file) => {
    return file.split('/').length > 3
  })
  /**
   * only keep the big versions (like vX.Y) and the latest smaller version (like vX.Y.Z) that if without corresponding big version
   * for example:
   * input: ["xxx/xxx/v2.1.0/xxx.md", "xxx/xxx/v2.1.1/xxx.md", "xxx/xxx/v2.2.1/xxx.md", "xxx/xxx/v2.2.2/xxx.md"]
   * output: ["xxx/xxx/v2.1.1/xxx.md", "xxx/xxx/v2.2.2/xxx.md"]
   * the smaller versions "xxx/xxx/v2.2.0.1" will be filtered because it is lower than "v2.2.0.2"
   */
  let filesGroup = lodash.groupBy(files, (file) => {
    let [category, name, version] = file.split('/')
    if (version.split('.').length > 2) {
      const [v1, v2] = version.split('.')
      version = `${v1}.${v2}`
    }
    return `${category}-${name}-${version}`
  })
  files = Object.values(filesGroup).map((group) => {
    return group.sort((a, b) => {
      let [, , aVersion] = a.split('/')
      let [, , bVersion] = b.split('/')
      try {
        return cv.compareVersions(bVersion, aVersion)
      } catch (error) {
        return 1
      }
    })[0]
  })
  // only get the latest version
  filesGroup = lodash.groupBy(files, (file) => {
    let [category, name] = file.split('/')
    return `${category}-${name}`
  })
  const _files = Object.values(filesGroup).map((group) => {
    return group
      .sort((a, b) => {
        let [, , aVersion] = a.split('/')
        let [, , bVersion] = b.split('/')
        try {
          return cv.compareVersions(bVersion, aVersion)
        } catch (error) {
          return 1
        }
      })
      .slice(0, Math.min(1, group.length))
  })
  return lodash.flatten(_files)
}

async function addRedirect(dir) {
  dir = dir.slice(-1) === '/' ? dir : dir + '/'
  const files = await getLatestVersionFiles(dir)

  /**
   * redirects:
   *  - /category/name
   */
  for (let file of files) {
    let [category, name] = file.split('/')
    category = category.trim().toLowerCase().endsWith('s')
      ? category.substring(0, category.length - 1)
      : category

    const mdpath = dir + file
    let data = fs.readFileSync(mdpath, 'utf8')

    let yamlFile = path.join(
      path.dirname(mdpath),
      '..',
      path.basename(mdpath, '.md') + '.yaml'
    )
    let ymlData = null
    if (fs.existsSync(yamlFile)) {
      ymlData = yaml.load(fs.readFileSync(yamlFile, 'utf8'))
    }
    if (ymlData) {
      if (ymlData.icon) {
        data = data.replace(/icon: .*/, 'icon: ' + ymlData.icon)
      }
      if (ymlData.sn_available) {
        data = data.replace(/(sn_available:.*\n)/, '')
        data = data.replace(
          /(id:.*\n---)/,
          'sn_available: ' + ymlData.sn_available + '\n$1'
        )
      }
    }
    // check if `/(id:.*\n---)/g` is exist in data
    if (data.match(/(id:.*\n---)/g)) {
      data = data.replace(
        /(id:.*\n---)/g,
        'redirects:\n - /hub/' + category + '-' + name + '\n$1'
      )
    } else if (data.match(/(powered_by:.*\n---)/g)) {
      // check if `/(powered_by:.*\n---)/g` is exist in data
      data = data.replace(
        /(powered_by:.*\n---)/g,
        'redirects:\n - /hub/' + category + '-' + name + '\n$1'
      )
    }
    fs.writeFileSync(dir + file, data)
    console.log(
      dir +
        file +
        ' add redirects to ' +
        category +
        '-' +
        name +
        ' successfully'
    )
  }
}

// Example: node fix-hub-redirect.js connectors/aws-lambda-sink/2.10.0/aws-lambda-sink.md
const [dir] = process.argv.slice(2)
addRedirect(path.join(___dirname, dir))

// Usage:
// Example 1: Update a specify md file:
// node fix-hub-redirect.js connectors/aws-lambda-sink/2.10.0/aws-lambda-sink.md

// Example 2: Update files in a specify directory:
// node fix-hub-redirect.js connectors/aws-lambda-sink

// Example 3: Update all connectors docs:
// node fix-hub-redirect.js connectors

// Example 4: Update all docs:
// node fix-hub-redirect.js .


const fs = require('fs')
const path = require('path')
const ___dirname = path.join(__dirname, '../..')

function travel(dir, callback) {
  if (dir.includes('node_modules')) {
    return
  }
  if (fs.statSync(dir).isFile()) {
    if (dir.endsWith('.md')) {
      callback(dir)
    }
  } else {
    fs.readdirSync(dir).forEach((file) => {
      var pathname = path.join(dir, file)
      if (fs.statSync(pathname).isDirectory()) {
        travel(pathname, callback)
      } else if (pathname.endsWith('.md')) {
        callback(pathname)
      }
    })
  }
}

function fix(mdpath) {
  let data = fs.readFileSync(mdpath, 'utf8')
  let updateData = fs.readFileSync(mdpath, 'utf8')
  let reg = /([ ]*)> \*+Note.*\n(([ ]*>\s*.*\n)*)/g
  while ((m = reg.exec(data))) {
    updateData = updateData.replace(
      m[0],
      m[1] +
        '{% callout title="Note" type="note" %}\n' +
        m[2].replace(/>[ \n]/g, '') +
        m[1] +
        '{% /callout %}\n'
    )
  }
  reg = /([ ]*)> \*+Tip.*\n(([ ]*>\s*.*\n)*)/g
  while ((m = reg.exec(data))) {
    updateData = updateData.replace(
      m[0],
      m[1] +
        '{% callout title="Tip" type="tip" %}\n' +
        m[2].replace(/>[ \n]/g, '') +
        m[1] +
        '{% /callout %}\n'
    )
  }
  fs.writeFileSync(mdpath, updateData)
  console.log(mdpath + ' fixed')
}

// Example: node fix-hub-note.js connectors/aws-lambda-sink/2.10.0/aws-lambda-sink.md
const [dir] = process.argv.slice(2)
travel(path.join(___dirname, dir), fix)

// Usage:
// Example 1: Update a specify md file:
// node fix-hub-note.js connectors/aws-lambda-sink/2.10.0/aws-lambda-sink.md

// Example 2: Update files in a specify directory:
// node fix-hub-note.js connectors/aws-lambda-sink

// Example 3: Update all connectors docs:
// node fix-hub-note.js connectors

// Example 4: Update all docs:
// node fix-hub-note.js .


const fs = require('fs')
const path = require('path')
const ___dirname = path.join(__dirname, '../..')

function travel(dir, callback) {
  if (fs.statSync(dir).isFile()) {
    if (dir.endsWith('.md')) {
      callback(dir)
    }
  } else {
    fs.readdirSync(dir).forEach((file) => {
      var pathname = path.join(dir, file)
      if (fs.statSync(pathname).isDirectory()) {
        travel(pathname, callback)
      } else if (pathname.endsWith('.md')) {
        callback(pathname)
      }
    })
  }
}

function fix(mdpath) {
  let data = fs.readFileSync(mdpath, 'utf8')
  data = data
    .replace(/::: tabs/g, '{% tabs %}')
    .replace(/:::/g, '{% /tabs %}')
    .replace(/@@@[ ]+(?<tab_name>.+)/g, '{% tab label="$<tab_name>" %}')
    .replace(/@@@\s*\n/g, '{% /tab %}\n')
  fs.writeFileSync(mdpath, data)
  console.log(mdpath + ' fixed')
}

// Example: node fix-hub-tab.js connectors/aws-lambda-sink/2.10.0/aws-lambda-sink.md
const [dir] = process.argv.slice(2)
travel(path.join(___dirname, dir), fix)

// Usage:
// Example 1: Update a specify md file:
// node fix-hub-tab.js connectors/aws-lambda-sink/2.10.0/aws-lambda-sink.md

// Example 2: Update files in a specify directory:
// node fix-hub-tab.js connectors/aws-lambda-sink

// Example 3: Update all connectors docs:
// node fix-hub-tab.js connectors

// Example 4: Update all docs:
// node fix-hub-tab.js .


Before you begin

Steps

Kafka Python

Connect to your cluster using the Kafka Python client

StreamNative Documentation

Learn how to get started with StreamNative Cloud, Private Cloud, and our comprehensive data streaming platform.

Build with StreamNative

Overview

Welcome to StreamNative Cloud

Concepts

StreamNative Cloud Concepts

Engines

Data Streaming Engine

Quick Start

Quick Start for StreamNative Cloud

Kafka APIs

QuickStart - Kafka Client

Cloud Console

StreamNative Cloud Console Basics

BYOC Quick Start

Quick Start for StreamNative BYOC

Ursa Engine Quick Start

Quick Start for Ursa-Engine BYOC Cluster

StreamNative Cloud Clusters Overview

Cluster Types and Regions

Cluster Types & Regions in StreamNative Cloud

Release Channel

Geo Replication

Geo Replication on StreamNative Cloud

Maintenance Window

Manage Data Streams in StreamNative Cloud

Work with Tenants

Work with Tenants in StreamNative Cloud

Work with Namespaces

Work with Namespaces in StreamNative Cloud

Work with Topics

Work with Topics in StreamNative Cloud

Manage Security in StreamNative Cloud

Manage Secrets

Work with Secrets

Data Governance on StreamNative Cloud

Pulsar IO Connectors

Features

Feature Overview For Lakehouse Storage Support

Networking in StreamNative Cloud

Test Connectivity

Test Connectivity to StreamNative Cloud

Metrics API

Cluster Metrics

Advanced Observability

Advanced Observability Integration

Manage Notifications

Log Console

Grafana Dashboards and Alerting rules

Manage Universal Linking in StreamNative Cloud

Create Universial Links

Create Universal Links in StreamNative Cloud

Manage Universal Links

Manage Universal Links in StreamNative Cloud

StreamNative Cloud Billing

Manage Billing - UI

Manage billing using StreamNative Cloud Console

View Your Usage - UI

View your usage using StreamNative Cloud Console

AWS Marketplace - Pay As You Go

Get started with StreamNative Cloud on AWS Marketplace with Pay-As-You-Go

AWS Marketplace with Commitments

Get started with StreamNative Cloud on AWS Marketplace with commitments

Google Cloud Marketplace - Pay As You Go

Get started with StreamNative Cloud on Google Cloud Marketplace with Pay-As-You-Go

Google Cloud Marketplace with Commitments

Get started with StreamNative Cloud on Google Cloud Marketplace with commitments

Azure Cloud Marketplace - Pay As You Go

Get started with StreamNative Cloud on Azure Cloud Marketplace with Pay-As-You-Go

Azure Cloud Marketplace with Commitments

Get started with StreamNative Cloud on Azure Cloud Marketplace with commitments

Stop Charges for Your Pulsar Cluster

Stop charges for your Pulsar cluster

Billing FAQ

Generate a HAR file for Troubleshooting

StreamNative Cloud Objects

Understanding StreamNative Cloud objects

Glossary

Glossary of Terms & Concepts

Discover and deploy connectors to integrate StreamNative with hundreds of data sources and sinks.

StreamNative Connector Hub

ActiveMQ Connector integrates Apache Pulsar with Apache ActiveMQ.

ActiveMQ

support sink/source for AMQP version 1.0.0

AMQP 1 0

The Canal source connector pulls messages from MySQL to Pulsar topics.

Canal

The Debezium MongoDB source connector pulls messages from MongoDB and persists the messages to Pulsar topics

Debezium MongoDB

The Debezium Microsoft SQL Server source connector pulls messages from SQL Server and persists the messages to Pulsar topics

Debezium MSSQL

The Debezium source connector pulls messages from MySQL and persists the messages to Pulsar topics.

Debezium MySQL

The Debezium source connector pulls messages from PostgreSQL and persists the messages to Pulsar topics.

Debezium PostgreSQL

The File source connector pulls messages from files in directories and persists the messages to Pulsar topics.

File

The Flume source connector pulls messages from logs to Pulsar topics.

Flume

BigQuery Connector integrates Apache Pulsar with Google BigQuery.

Google BigQuery

The Google Pub/Sub sink connector allows you to write messages from Google Pub/Sub to Apache Pulsar.

Google Pub/Sub

JR Source Connector for Apache Kafka Connect. JR (jrnd.io) is a CLI program that helps you to stream quality random data for your applications.

Kafka Connect JR

The official MongoDB Kafka Connect Source connector.

Kafka Connect MongoDB

The official YugabyteDB CDC Kafka Source Connector for capturing change data from YugabyteDB.

Kafka Connect YugabyteDB CDC

The Kafka source connector pulls messages from Kafka topics and persists the messages to Pulsar topics.

Kafka

The Kinesis source connector pulls data from Amazon Kinesis and persists data into Pulsar

Kinesis

Lakehouse

The Netty source connector opens a port that accepts incoming data via the configured network protocol and publish it to user-defined Pulsar topics

Netty

The RabbitMQ source connector receives messages from RabbitMQ clusters and writes messages to Pulsar topics

RabbitMQ

The SQS source connector is used to consume messages from Amazon SQS and publish them to Pulsar.

The Twitter Firehose source connector receives tweets from Twitter Firehose and writes the tweets to Pulsar topics.

Twitter Firehose

The Aerospike sink connector pulls messages from Pulsar topics to Aerospike clusters

Aerospike

This connector allows you to make sink connections from Pulsar to AWS EventBridge.

AWS EventBridge

The AWS Lambda sink connector allows you to send messages from Apache Pulsar to AWS Lambda.

AWS Lambda

Cloud Storage Connector integrates Apache Pulsar with cloud storage.

AWS S3

Azure Blob Storage

The Cassandra sink connector pulls messages from Pulsar topics to Cassandra clusters

Cassandra

The Elasticsearch sink connector pulls messages from Pulsar topics and persists the messages to indexes

Elasticsearch

The Flume sink connector pulls messages from Pulsar topics to logs

Google Cloud Storage

The Google Pub/Sub sink connector is used to write messages from Apache Pulsar topics to Google Cloud Pub/Sub.

The HBase sink connector pulls the messages from Pulsar topics and persists the messages to HBase tables

HBase

The HDFS3 sink connector pulls the messages from Pulsar topics and persists the messages to HDFS files.

HDFS3

The InfluxDB sink connector pulls messages from Pulsar topics and persists the messages to InfluxDB.

InfluxDB

The JDBC sink connector pulls messages from Pulsar topics and persists the messages to MySQL or SQlite.

JDBC ClickHouse

JDBC MariaDB

JDBC PostgreSQL

JDBC SQLite

kafka-connect-bigquery is a Kafka Connect connector for Google BigQuery.

Kafka Connect BigQuery

kafka-connect-datagen is a Kafka Connect connector for generating mock data for testing and is not suitable for production scenarios.

Kafka Connect Datagen

The official ElasticSearch Kafka Connect Sink connector.

Kafka Connect Elasticsearch

The official Apache Iceberg Kafka Connect Sink connector.

Kafka Connect Iceberg

The official Milvus Kafka Connect Sink connector.

Kafka Connect Milvus

The official MongoDB Kafka Connect Sink connector.

The official Snowflake Kafka Connect Sink connector.

Kafka Connect Snowflake

The Kafka sink connector pulls messages from Pulsar topics and persists the messages to Kafka topics.

The Kinesis sink connector pulls data from Pulsar and persists data into Amazon Kinesis.

The MongoDB sink connector pulls messages from Pulsar topics and persists the messages to collections.

MongoDB

Pinecone

The RabbitMQ sink connector pulls messages from Pulsar topics and persist the messages to RabbitMQ queues.

The Redis sink connector pulls messages from Pulsar topics and persists the messages to a Redis database.

Redis

The Snowflake sink connector is used to write messages from Apache Pulsar topics to Snowflake tables.

Snowflake

The Snowflake sink connector is used to write messages from Apache Pulsar topics to Snowflake tables with snowpipe streaming.

Snowflake Streaming

The Solr sink connector pulls messages from Pulsar topics and persists the messages to Solr collections.

Solr

The SQS source connector is used to consume messages from Pulsar topics and publish them to AWS SQS.

Kafka Clients

Pulsar Clients

MQTT Clients

Connect to your cluster using the Kafka Python client

Before you begin

Steps

Kafka Clients

Pulsar Clients

MQTT Clients

​Before you begin

​Steps

Before you begin

Steps