diff --git a/scripts/integration/kafka/Cargo.lock b/scripts/integration/kafka/Cargo.lock
new file mode 100644
index 0000000000000..d6b4902461d60
--- /dev/null
+++ b/scripts/integration/kafka/Cargo.lock
@@ -0,0 +1,252 @@
+# This file is automatically @generated by Cargo.
+# It is not intended for manual editing.
+version = 3
+
+[[package]]
+name = "anyhow"
+version = "1.0.66"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "216261ddc8289130e551ddcd5ce8a064710c0d064a4d2895c67151c92b5443f6"
+
+[[package]]
+name = "atty"
+version = "0.2.14"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d9b39be18770d11421cdb1b9947a45dd3f37e93092cbf377614828a319d5fee8"
+dependencies = [
+ "hermit-abi",
+ "libc",
+ "winapi",
+]
+
+[[package]]
+name = "bitflags"
+version = "1.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
+
+[[package]]
+name = "clap"
+version = "4.0.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "335867764ed2de42325fafe6d18b8af74ba97ee0c590fa016f157535b42ab04b"
+dependencies = [
+ "atty",
+ "bitflags",
+ "clap_derive",
+ "clap_lex",
+ "once_cell",
+ "strsim",
+ "termcolor",
+]
+
+[[package]]
+name = "clap_derive"
+version = "4.0.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "16a1b0f6422af32d5da0c58e2703320f379216ee70198241c84173a8c5ac28f3"
+dependencies = [
+ "heck",
+ "proc-macro-error",
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "clap_lex"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0d4198f73e42b4936b35b5bb248d81d2b595ecb170da0bac7655c54eedfa8da8"
+dependencies = [
+ "os_str_bytes",
+]
+
+[[package]]
+name = "dunce"
+version = "1.0.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0bd4b30a6560bbd9b4620f4de34c3f14f60848e58a9b7216801afcb4c7b31c3c"
+
+[[package]]
+name = "heck"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2540771e65fc8cb83cd6e8a237f70c319bd5c29f78ed1084ba5d50eeac86f7f9"
+
+[[package]]
+name = "hermit-abi"
+version = "0.1.19"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "62b467343b94ba476dcb2500d242dadbb39557df889310ac77c5d99100aaac33"
+dependencies = [
+ "libc",
+]
+
+[[package]]
+name = "itoa"
+version = "1.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4217ad341ebadf8d8e724e264f13e593e0648f5b3e94b3896a5df283be015ecc"
+
+[[package]]
+name = "libc"
+version = "0.2.137"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fc7fcc620a3bff7cdd7a365be3376c97191aeaccc2a603e600951e452615bf89"
+
+[[package]]
+name = "once_cell"
+version = "1.16.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "86f0b0d4bf799edbc74508c1e8bf170ff5f41238e5f8225603ca7caaae2b7860"
+
+[[package]]
+name = "os_str_bytes"
+version = "6.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3baf96e39c5359d2eb0dd6ccb42c62b91d9678aa68160d261b9e0ccbf9e9dea9"
+
+[[package]]
+name = "proc-macro-error"
+version = "1.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "da25490ff9892aab3fcf7c36f08cfb902dd3e71ca0f9f9517bea02a73a5ce38c"
+dependencies = [
+ "proc-macro-error-attr",
+ "proc-macro2",
+ "quote",
+ "syn",
+ "version_check",
+]
+
+[[package]]
+name = "proc-macro-error-attr"
+version = "1.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a1be40180e52ecc98ad80b184934baf3d0d29f979574e439af5a55274b35f869"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "version_check",
+]
+
+[[package]]
+name = "proc-macro2"
+version = "1.0.47"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5ea3d908b0e36316caf9e9e2c4625cdde190a7e6f440d794667ed17a1855e725"
+dependencies = [
+ "unicode-ident",
+]
+
+[[package]]
+name = "quote"
+version = "1.0.21"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bbe448f377a7d6961e30f5955f9b8d106c3f5e449d493ee1b125c1d43c2b5179"
+dependencies = [
+ "proc-macro2",
+]
+
+[[package]]
+name = "ryu"
+version = "1.0.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4501abdff3ae82a1c1b477a17252eb69cee9e66eb915c1abaa4f44d873df9f09"
+
+[[package]]
+name = "serde"
+version = "1.0.147"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d193d69bae983fc11a79df82342761dfbf28a99fc8d203dca4c3c1b590948965"
+
+[[package]]
+name = "serde_json"
+version = "1.0.87"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6ce777b7b150d76b9cf60d28b55f5847135a003f7d7350c6be7a773508ce7d45"
+dependencies = [
+ "itoa",
+ "ryu",
+ "serde",
+]
+
+[[package]]
+name = "strsim"
+version = "0.10.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "73473c0e59e6d5812c5dfe2a064a6444949f089e20eec9a2e5506596494e4623"
+
+[[package]]
+name = "syn"
+version = "1.0.103"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a864042229133ada95abf3b54fdc62ef5ccabe9515b64717bcb9a1919e59445d"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "unicode-ident",
+]
+
+[[package]]
+name = "termcolor"
+version = "1.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bab24d30b911b2376f3a13cc2cd443142f0c81dda04c118693e35b3835757755"
+dependencies = [
+ "winapi-util",
+]
+
+[[package]]
+name = "unicode-ident"
+version = "1.0.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6ceab39d59e4c9499d4e5a8ee0e2735b891bb7308ac83dfb4e80cad195c9f6f3"
+
+[[package]]
+name = "vector-kafka-env-manager"
+version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "clap",
+ "dunce",
+ "serde_json",
+]
+
+[[package]]
+name = "version_check"
+version = "0.9.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "49874b5167b65d7193b8aba1567f5c7d93d001cafc34600cee003eda787e483f"
+
+[[package]]
+name = "winapi"
+version = "0.3.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419"
+dependencies = [
+ "winapi-i686-pc-windows-gnu",
+ "winapi-x86_64-pc-windows-gnu",
+]
+
+[[package]]
+name = "winapi-i686-pc-windows-gnu"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6"
+
+[[package]]
+name = "winapi-util"
+version = "0.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "70ec6ce85bb158151cae5e5c87f95a8e97d2c0c4b001223f33a334e3ce5de178"
+dependencies = [
+ "winapi",
+]
+
+[[package]]
+name = "winapi-x86_64-pc-windows-gnu"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"
diff --git a/scripts/integration/kafka/Cargo.toml b/scripts/integration/kafka/Cargo.toml
new file mode 100644
index 0000000000000..19c38b082de63
--- /dev/null
+++ b/scripts/integration/kafka/Cargo.toml
@@ -0,0 +1,16 @@
+[package]
+name = "vector-kafka-env-manager"
+version = "0.1.0"
+edition = "2021"
+authors = ["Vector Contributors <vector@datadoghq.com>"]
+license = "MPL-2.0"
+readme = "README.md"
+publish = false
+
+[dependencies]
+anyhow = "1.0.66"
+clap = { version = "4.0.18", features = ["derive"] }
+dunce = "1.0.3"
+serde_json = "1.0.87"
+
+[workspace]
diff --git a/scripts/integration/kafka/README.md b/scripts/integration/kafka/README.md
new file mode 100644
index 0000000000000..fb6299563bd0e
--- /dev/null
+++ b/scripts/integration/kafka/README.md
@@ -0,0 +1,11 @@
+# vector-kafka-env-manager
+
+-----
+
+## Usage
+
+```text
+vdev int show kafka
+vdev int start kafka <ENV>
+vdev int stop kafka <ENV>
+```
diff --git a/scripts/integration/kafka/data/docker-compose.yml b/scripts/integration/kafka/data/docker-compose.yml
new file mode 100644
index 0000000000000..3c1af7dbd8774
--- /dev/null
+++ b/scripts/integration/kafka/data/docker-compose.yml
@@ -0,0 +1,40 @@
+version: '3'
+
+services:
+  zookeeper:
+    image: docker.io/wurstmeister/zookeeper:${KAFKA_VERSION}
+    ports:
+    - 2181:2181
+  kafka:
+    image: docker.io/wurstmeister/kafka:2.13-2.6.0
+    depends_on:
+    - zookeeper
+    environment:
+    - KAFKA_BROKER_ID=1
+    - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
+    - KAFKA_LISTENERS=PLAINTEXT://:9091,SSL://:9092,SASL_PLAINTEXT://:9093
+    - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9091,SSL://kafka:9092,SASL_PLAINTEXT://kafka:9093
+    - KAFKA_SSL_KEYSTORE_TYPE=PKCS12
+    - KAFKA_SSL_KEYSTORE_LOCATION=/certs/kafka.p12
+    - KAFKA_SSL_KEYSTORE_PASSWORD=NOPASS
+    - KAFKA_SSL_TRUSTSTORE_TYPE=PKCS12
+    - KAFKA_SSL_TRUSTSTORE_LOCATION=/certs/kafka.p12
+    - KAFKA_SSL_TRUSTSTORE_PASSWORD=NOPASS
+    - KAFKA_SSL_KEY_PASSWORD=NOPASS
+    - KAFKA_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM=none
+    - KAFKA_OPTS=-Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf
+    - KAFKA_INTER_BROKER_LISTENER_NAME=SASL_PLAINTEXT
+    - KAFKA_SASL_ENABLED_MECHANISMS=PLAIN
+    - KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL=PLAIN
+    ports:
+    - 9091:9091
+    - 9092:9092
+    - 9093:9093
+    volumes:
+    - ../../../../tests/data/ca/intermediate_server/private/kafka.p12:/certs/kafka.p12:ro
+    - ../../../../tests/data/kafka_server_jaas.conf:/etc/kafka/kafka_server_jaas.conf
+
+networks:
+  default:
+    name: ${VECTOR_NETWORK}
+    external: true
diff --git a/scripts/integration/kafka/src/core.rs b/scripts/integration/kafka/src/core.rs
new file mode 100644
index 0000000000000..0dd4e1fa31a2a
--- /dev/null
+++ b/scripts/integration/kafka/src/core.rs
@@ -0,0 +1,61 @@
+use anyhow::{bail, Result};
+use serde_json::Value;
+use std::path::PathBuf;
+use std::process::Command;
+use std::thread;
+use std::time::Duration;
+
+pub fn start(config: Value) -> Result<()> {
+    let mut command = compose_command();
+    command.args(["up", "-d"]);
+
+    apply_env_vars(&mut command, &config);
+
+    let status = command.status()?;
+    if status.success() {
+        thread::sleep(Duration::from_secs(20));
+        return Ok(());
+    } else {
+        bail!("failed to execute: {}", render_command(&mut command));
+    }
+}
+
+pub fn stop(config: Value) -> Result<()> {
+    let mut command = compose_command();
+    command.args(["down", "-t", "0"]);
+
+    apply_env_vars(&mut command, &config);
+
+    let status = command.status()?;
+    if status.success() {
+        return Ok(());
+    } else {
+        bail!("failed to execute: {}", render_command(&mut command));
+    }
+}
+
+fn compose_command() -> Command {
+    let path = PathBuf::from_iter(["data", "docker-compose.yml"].iter());
+    let compose_file = match dunce::canonicalize(&path) {
+        Ok(p) => p.display().to_string(),
+        Err(_) => path.display().to_string(),
+    };
+
+    let mut command = Command::new("docker");
+    command.args(["compose", "-f", &compose_file]);
+    command
+}
+
+fn apply_env_vars(command: &mut Command, config: &Value) {
+    if let Some(version) = config.get("version") {
+        command.env("KAFKA_VERSION", version.as_str().unwrap());
+    }
+}
+
+fn render_command(command: &mut Command) -> String {
+    format!(
+        "{} {}",
+        command.get_program().to_str().unwrap(),
+        Vec::from_iter(command.get_args().map(|arg| arg.to_str().unwrap())).join(" ")
+    )
+}
diff --git a/scripts/integration/kafka/src/main.rs b/scripts/integration/kafka/src/main.rs
new file mode 100644
index 0000000000000..c063207228ab4
--- /dev/null
+++ b/scripts/integration/kafka/src/main.rs
@@ -0,0 +1,26 @@
+mod core;
+
+use anyhow::Result;
+use clap::{Parser, Subcommand};
+
+#[derive(Parser, Debug)]
+#[command(disable_help_subcommand = true)]
+struct Cli {
+    #[command(subcommand)]
+    command: Commands,
+}
+
+#[derive(Subcommand, Debug)]
+enum Commands {
+    Start { json: String },
+    Stop { json: String },
+}
+
+fn main() -> Result<()> {
+    let cli = Cli::parse();
+
+    match &cli.command {
+        Commands::Start { json } => core::start(serde_json::from_str(&json)?),
+        Commands::Stop { json } => core::stop(serde_json::from_str(&json)?),
+    }
+}
diff --git a/scripts/integration/kafka/test.yaml b/scripts/integration/kafka/test.yaml
new file mode 100644
index 0000000000000..92b198bf327e0
--- /dev/null
+++ b/scripts/integration/kafka/test.yaml
@@ -0,0 +1,11 @@
+args:
+- --features
+- kafka-integration-tests
+- --lib
+- '::kafka::'
+
+env:
+  KAFKA_HOST: kafka
+
+matrix:
+- version: latest