How to Monitor Apache Bahir with Vigilmon (Uptime, APIs, and Alerts)

Apache Bahir provides streaming connectors and SQL extensions for Apache Spark and Apache Flink, bridging the gap between those processing engines and external data sources like MQTT brokers, Twitter Firehose, Slack, Kinesis, and CouchDB. Bahir connectors live at the edge of your streaming pipeline — they are the first thing to break when an upstream broker goes down or a network partition isolates your cluster from its data sources.

When your MQTT connector loses its broker connection, Bahir silently buffers events until it exhausts memory. When the Slack event connector fails to authenticate, your Flink job continues running but ingests nothing — and without external monitoring, you have no signal that your data pipeline is starving. This tutorial shows you how to set up uptime monitoring for Apache Bahir deployments using Vigilmon — free tier, no credit card.

Why Bahir deployments need external monitoring

Bahir adds monitoring surface area at every external connector boundary:

MQTT broker disconnects — if the MQTT broker (Mosquitto, HiveMQ, EMQ X) becomes unreachable, Bahir's MQTT connector attempts to reconnect on an exponential backoff, but the Spark or Flink job continues running and reporting healthy while ingesting zero messages
OAuth token expiry — connectors for Twitter and Slack use OAuth credentials that can be revoked or expired; when this happens the connector silently stops delivering events rather than crashing
Broker authentication failures — a credential rotation on the upstream broker side causes connector reconnection to fail permanently until the Bahir job is restarted with new credentials
Spark Streaming batch accumulation — if Bahir's Spark DStream falls behind, micro-batches accumulate in the receiver; without external monitoring you won't notice until memory pressure kills the executor
Flink checkpoint failures — Bahir's Flink connectors rely on Flink's checkpointing for fault tolerance; if checkpoints start failing, a job restart will replay from an outdated offset, causing duplicate or missed records

External monitoring from Vigilmon gives you real-time visibility across all these layers.

What you'll need

A running Bahir connector deployment on Spark Streaming or Flink
Access to the underlying platform's management endpoints
A free Vigilmon account (sign up takes 30 seconds)

Step 1: Expose health endpoints for your Bahir pipeline

Bahir does not ship HTTP health endpoints, but the platforms and brokers it connects to do. Verify these are accessible:

# Spark master web UI
curl http://spark-master.example.com:8080/

# Flink Job Manager REST API
curl http://flink-jobmanager.example.com:8081/jobs/overview

# MQTT broker health (HiveMQ)
curl http://hivemq.example.com:8080/api/v1/health

# MQTT broker health (Mosquitto via mosquitto_pub)
mosquitto_pub -h mqtt.example.com -p 1883 -t '$SYS/broker/uptime' -n

Add a sidecar health endpoint to your Bahir application that verifies connectivity to the upstream broker:

# Lightweight sidecar that checks MQTT broker reachability
import paho.mqtt.client as mqtt
from flask import Flask, jsonify
import threading, time

app = Flask(__name__)
_mqtt_ok = {"value": False, "checked_at": 0}

def check_mqtt():
    while True:
        client = mqtt.Client()
        try:
            client.connect("mqtt.example.com", 1883, keepalive=5)
            client.loop_start()
            time.sleep(2)
            client.loop_stop()
            client.disconnect()
            _mqtt_ok["value"] = True
        except Exception:
            _mqtt_ok["value"] = False
        _mqtt_ok["checked_at"] = time.time()
        time.sleep(30)

@app.get("/health")
def health():
    lag = time.time() - _mqtt_ok["checked_at"]
    if not _mqtt_ok["value"] or lag > 90:
        return jsonify({"status": "degraded", "mqtt_reachable": _mqtt_ok["value"]}), 503
    return jsonify({"status": "ok", "mqtt_reachable": True}), 200

threading.Thread(target=check_mqtt, daemon=True).start()
app.run(host="0.0.0.0", port=8080)

For Docker deployments:

services:
  bahir-health-sidecar:
    image: your-org/bahir-health-sidecar:latest
    ports:
      - "8080:8080"
    environment:
      MQTT_HOST: mqtt.example.com
      MQTT_PORT: "1883"
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8080/health"]
      interval: 30s
      timeout: 10s
      retries: 3
    restart: unless-stopped

Step 2: Monitor the MQTT broker health endpoint

Log in to vigilmon.online and go to Monitors → New Monitor
Choose HTTP / HTTPS as the type
URL: http://hivemq.example.com:8080/api/v1/health (or your broker's health path)
Set check interval to 1 minute
Under Expected response, configure:
- Status code: 200
Save the monitor

This catches broker outages independently of the Bahir connector — if the broker fails, your Bahir job loses its data source regardless of connector health.

Step 3: Monitor the Spark or Flink platform

If Bahir runs on Spark Streaming:

Go to Monitors → New Monitor
Choose HTTP / HTTPS
URL: http://spark-master.example.com:8080/
Expected status: 200
Response body contains: Spark Master
Save the monitor

If Bahir runs on Flink:

Go to Monitors → New Monitor
Choose HTTP / HTTPS
URL: http://flink-jobmanager.example.com:8081/jobs/overview
Expected status: 200
Response body contains: "jobs"
Save the monitor

Step 4: Add TCP port monitoring for brokers and platforms

TCP checks confirm that the broker port itself is open, independent of any HTTP management layer:

Go to Monitors → New Monitor
Choose TCP Port
Host: mqtt.example.com, Port: 1883
Save the monitor

| Service | Default port | |---------|-------------| | MQTT (unencrypted) | 1883 | | MQTT over TLS | 8883 | | Spark Master RPC | 7077 | | Flink Job Manager RPC | 6123 | | Flink REST | 8081 | | ZooKeeper | 2181 |

For production MQTT deployments using TLS (port 8883), add a TCP check on that port too and enable SSL certificate monitoring so you know before the cert expires.

Step 5: Set up a heartbeat monitor for your Bahir pipeline output

If your Bahir pipeline writes to a database, file store, or downstream API, instrument your output writer to ping a heartbeat URL whenever it successfully writes a batch:

Go to Monitors → New Monitor
Choose Heartbeat
Name it bahir-pipeline-output-heartbeat
Set the expected interval to 2x your micro-batch interval (e.g., if batches are every 30 seconds, set heartbeat to 2 minutes)
Copy the generated heartbeat URL
Add the ping to your output writer:

import requests, os

HEARTBEAT_URL = os.environ["VIGILMON_HEARTBEAT_URL"]

def write_batch(records):
    # ... your write logic ...
    db.bulk_insert(records)
    requests.get(HEARTBEAT_URL, timeout=5)  # ping on success

If the Bahir connector stops delivering records or your writer crashes, the heartbeat stops arriving and Vigilmon opens an incident.

Step 6: Configure webhook alerts

Bahir connector failures are silent by default — the job stays green in the platform UI while ingesting nothing. Webhook alerts from Vigilmon are your only real-time signal.

Go to Alert Channels → Add Channel → Webhook
Enter your Slack, PagerDuty, or OpsGenie webhook URL
Assign the channel to all your Bahir monitors

Example alert payload from Vigilmon:

{
  "monitor_name": "MQTT Broker TCP 1883",
  "status": "down",
  "host": "mqtt.example.com",
  "port": 1883,
  "started_at": "2026-05-20T03:14:00Z",
  "duration_seconds": 90
}

Configure different alert channels for broker owners and pipeline owners — they need different response runbooks when an MQTT broker goes down.

Step 7: Create a status page for your streaming pipeline

If downstream teams consume data from your Bahir pipeline, give them a status page so they can self-serve status checks:

Go to Status Pages → New Status Page
Name it (e.g. "Streaming Data Pipeline Status")
Add all your monitors — broker HTTP, broker TCP, platform UI, heartbeat
Publish the page

Monitor coverage summary

| Monitor | Type | What it catches | |---------|------|-----------------| | http://hivemq.example.com:8080/api/v1/health | HTTP | MQTT broker outage | | mqtt.example.com:1883 | TCP | Broker port unreachable | | http://spark-master.example.com:8080/ | HTTP | Spark cluster failure | | http://flink-jobmanager.example.com:8081/jobs/overview | HTTP | Flink Job Manager failure | | bahir-pipeline-output-heartbeat | Heartbeat | Silent connector stall, no data flowing |

What's next

SSL certificate monitoring — if your MQTT broker uses TLS (port 8883), Vigilmon tracks the certificate expiry date and alerts before it lapses
Response time tracking — latency spikes on the broker health endpoint often predict imminent connection failures before they occur

Get started free at vigilmon.online — no credit card, monitors start running in under a minute.