Add pipeline broker support for the subject's internal pipeline bus

internal/config/case.go

@@ -52,6 +52,17 @@ type TestCase struct {
 	// kafka_performance / kafka_correctness types.
 	Kafka *KafkaConfig `yaml:"kafka"`
 
+	// PipelineBroker, when set, adds a SECOND, dedicated Redpanda broker
+	// (hostname "pipeline-broker") for the SUBJECT's own internal transport —
+	// e.g. VirtualMetric's `pipeline_bus: {type: kafka}`. It is independent of
+	// the `kafka:` block above (which feeds the generator / subject device):
+	// different hostname and port, so a case may use both at once with no
+	// collision. The subject depends_on it being healthy before starting, so the
+	// subject's broker dial can't race startup. Topics auto-create on first
+	// produce (the bus owns its topics). Point the subject config at
+	// "pipeline-broker:<port>" (default 19092).
+	PipelineBroker *PipelineBrokerConfig `yaml:"pipeline_broker"`
+
 	// Vault, when set, adds a HashiCorp Vault dev-mode server (TLS-enabled)
 	// to the test topology: the harness renders a `vault` service plus a
 	// one-shot `vault-init` that seeds the declared secrets, and the subject
@@ -292,6 +303,67 @@ func (k *KafkaConfig) SMPOrDefault() int {
 	return 1
 }
 
+// PipelineBrokerConfig configures the dedicated Redpanda broker the harness
+// stands up for the subject's internal pipeline bus (see TestCase.PipelineBroker).
+// All fields are optional; the orchestrator applies the defaults noted below.
+type PipelineBrokerConfig struct {
+	// Image is the Redpanda container image (default "redpandadata/redpanda:latest").
+	Image string `yaml:"image"`
+	// Memory is the Redpanda --memory allotment (default "1G").
+	Memory string `yaml:"memory"`
+	// SMP is the Redpanda --smp core count (default 1).
+	SMP int `yaml:"smp"`
+	// Port is the broker's advertised Kafka API port (default 19092). Kept
+	// distinct from the `kafka:` broker's 9092 so both can run side by side.
+	Port int `yaml:"port"`
+	// AutoCreate controls broker-side topic auto-creation. Unset/true is the
+	// Redpanda/Kafka convenience default; set false to mirror brokers that
+	// disallow it (e.g. Azure Event Hubs), in which case Topics MUST list the
+	// pipeline-bus topics to pre-create.
+	AutoCreate *bool `yaml:"auto_create"`
+	// Topics are pre-created by a one-shot (pipeline-broker-init) before the
+	// subject starts. Required when AutoCreate is false; the subject then
+	// depends_on that init completing rather than just the broker being healthy.
+	Topics []string `yaml:"topics"`
+}
+
+// AutoCreateOrDefault reports whether broker-side topic auto-creation is on
+// (default true).
+func (p *PipelineBrokerConfig) AutoCreateOrDefault() bool {
+	if p != nil && p.AutoCreate != nil {
+		return *p.AutoCreate
+	}
+	return true
+}
+
+func (p *PipelineBrokerConfig) ImageOrDefault() string {
+	if p != nil && p.Image != "" {
+		return p.Image
+	}
+	return "redpandadata/redpanda:latest"
+}
+
+func (p *PipelineBrokerConfig) MemoryOrDefault() string {
+	if p != nil && p.Memory != "" {
+		return p.Memory
+	}
+	return "1G"
+}
+
+func (p *PipelineBrokerConfig) SMPOrDefault() int {
+	if p != nil && p.SMP > 0 {
+		return p.SMP
+	}
+	return 1
+}
+
+func (p *PipelineBrokerConfig) PortOrDefault() int {
+	if p != nil && p.Port > 0 {
+		return p.Port
+	}
+	return 19092
+}
+
 // VaultConfig configures the in-topology HashiCorp Vault dev server (see
 // TestCase.Vault). All fields except Secrets are optional; the orchestrator
 // applies the defaults noted below. The dev server listens TLS-only at

internal/orchestrator/docker.go

@@ -94,7 +94,7 @@ services:
     image: "{{ .SubjectImage }}"
     container_name: "{{ .SubjectContainer }}"
     networks: [bench]
-{{- if or .KafkaEnabled .AWSEnabled .AzureEnabled .VaultEnabled .MinioEnabled }}
+{{- if or .KafkaEnabled .AWSEnabled .AzureEnabled .VaultEnabled .MinioEnabled .PipelineBrokerEnabled }}
     depends_on:
 {{- if .KafkaGSSAPIEnabled }}
       kafka-init:
@@ -119,6 +119,15 @@ services:
       minio-init:
         condition: service_completed_successfully
 {{- end }}
+{{- if .PipelineBrokerEnabled }}
+{{- if .PipelineBrokerHasTopics }}
+      pipeline-broker-init:
+        condition: service_completed_successfully
+{{- else }}
+      pipeline-broker:
+        condition: service_healthy
+{{- end }}
+{{- end }}
 {{- end }}
     volumes:
       - "{{ .ConfigSrc }}:{{ .ConfigDst }}{{ .ConfigMountOpts }}"
@@ -710,6 +719,59 @@ services:
 {{- end }}
     restart: "no"
 {{- end }}
+{{- if .PipelineBrokerEnabled }}
+
+  # Dedicated Redpanda broker for the SUBJECT's internal pipeline bus
+  # (e.g. VirtualMetric pipeline_bus type=kafka). Separate from the kafka block
+  # broker (hostname redpanda, port 9092) -- different hostname AND port
+  # (default 19092) so a case can use both at once without collision. The
+  # subject depends_on this being healthy, so its broker dial cannot race
+  # startup; topics auto-create on first produce (the bus owns its topics).
+  pipeline-broker:
+    image: "{{ .PipelineBrokerImage }}"
+    container_name: "bench-pipeline-broker"
+    hostname: "pipeline-broker"
+    networks: [bench]
+    command:
+      - redpanda
+      - start
+      - "--mode"
+      - "dev-container"
+      - "--smp"
+      - "{{ .PipelineBrokerSMP }}"
+      - "--memory"
+      - "{{ .PipelineBrokerMemory }}"
+      - "--kafka-addr"
+      - "PLAINTEXT://0.0.0.0:{{ .PipelineBrokerPort }}"
+      - "--advertise-kafka-addr"
+      - "PLAINTEXT://pipeline-broker:{{ .PipelineBrokerPort }}"
+      - "--set"
+      - "redpanda.auto_create_topics_enabled={{ .PipelineBrokerAutoCreate }}"
+    healthcheck:
+      test: ["CMD-SHELL", "rpk cluster health | grep -q 'Healthy:.*true'"]
+      interval: 3s
+      timeout: 5s
+      retries: 30
+      start_period: 5s
+    restart: "no"
+{{- if .PipelineBrokerHasTopics }}
+
+  # Pre-create the pipeline-bus topics before the subject starts -- mirrors a
+  # broker that disallows auto-create (e.g. Azure Event Hubs), where the bus
+  # topics must exist up front. The subject depends_on this completing.
+  pipeline-broker-init:
+    image: "{{ .PipelineBrokerImage }}"
+    container_name: "bench-pipeline-broker-init"
+    networks: [bench]
+    depends_on:
+      pipeline-broker:
+        condition: service_healthy
+    entrypoint: ["/bin/sh", "-c"]
+    command:
+      - "{{ .PipelineBrokerTopicCreateCmd }}"
+    restart: "no"
+{{- end }}
+{{- end }}
 {{- if .VaultEnabled }}
 
   vault:
@@ -1397,6 +1459,19 @@ type composeVars struct {
 	// the case actually uses.
 	KafkaBootstrapMechanism string
 
+	// PipelineBroker (a case's `pipeline_broker` block): a dedicated Redpanda
+	// ("pipeline-broker") for the SUBJECT's internal pipeline bus, separate from
+	// the `kafka:` broker (different hostname + port). The subject depends_on it
+	// being healthy before starting.
+	PipelineBrokerEnabled        bool
+	PipelineBrokerImage          string
+	PipelineBrokerMemory         string
+	PipelineBrokerSMP            int
+	PipelineBrokerPort           int
+	PipelineBrokerAutoCreate     bool
+	PipelineBrokerHasTopics      bool
+	PipelineBrokerTopicCreateCmd string
+
 	// Vault topology. VaultEnabled gates the vault + vault-init services,
 	// the subject's vault-init depends_on, and the /vault-tls subject mount.
 	// VaultInitCmd is the pre-built one-shot seeding shell line: mount and
@@ -1754,6 +1829,22 @@ func writeCompose(path string, cfg RunConfig) error {
 		vars.GenKafkaBatch = max(g.KafkaBatch, 1)
 	}
 
+	// Pipeline broker (a case's `pipeline_broker`): a dedicated Redpanda for the
+	// subject's internal pipeline bus, independent of the `kafka:` broker.
+	if tc.PipelineBroker != nil {
+		vars.PipelineBrokerEnabled = true
+		vars.PipelineBrokerImage = tc.PipelineBroker.ImageOrDefault()
+		vars.PipelineBrokerMemory = tc.PipelineBroker.MemoryOrDefault()
+		vars.PipelineBrokerSMP = tc.PipelineBroker.SMPOrDefault()
+		vars.PipelineBrokerPort = tc.PipelineBroker.PortOrDefault()
+		vars.PipelineBrokerAutoCreate = tc.PipelineBroker.AutoCreateOrDefault()
+		if len(tc.PipelineBroker.Topics) > 0 {
+			vars.PipelineBrokerHasTopics = true
+			vars.PipelineBrokerTopicCreateCmd = fmt.Sprintf("rpk topic create %s --brokers pipeline-broker:%d",
+				strings.Join(tc.PipelineBroker.Topics, " "), tc.PipelineBroker.PortOrDefault())
+		}
+	}
+
 	// Kafka (Redpanda) broker: render the redpanda + redpanda-init services and
 	// wire the generator/subject depends_on. Defaults centralized on KafkaConfig.
 	if tc.Kafka != nil {