Overview

Apache Kafka is a cornerstone for many real-time data pipelines, but managing its infrastructure can be complex.

Google Cloud Managed Service for Apache Kafka offers a fully managed solution, simplifying deployment and operations, however effective monitoring and management remain crucial for ensuring the health and performance of Kafka clusters.

This article provides a practical, step-by-step guide on setting up a Google Cloud Managed Service for Apache Kafka cluster and connecting it from Kpow using the OAUTHBEARER mechanism. We will walk through creating the necessary GCP resources, configuring a client virtual machine, and deploying a Kpow instance using Docker to demonstrate examples of monitoring and managing Kafka brokers and topics.

About Factor House

Factor House is a leader in real-time data tooling, empowering engineers with innovative solutions for Apache Kafka® and Apache Flink®.

Our flagship product, Kpow for Apache Kafka, is the market-leading enterprise solution for Kafka management and monitoring.

Explore our live multi-cluster demo environment or grab a free Community license and dive into streaming tech on your laptop with Factor House Local.

Create a Managed Kafka Cluster

We create GCP resources using the gcloud CLI. Once it is initialised, we should enable the Managed Kafka, Compute Engine, and Cloud DNS APIs as prerequisites.

gcloud services enable managedkafka.googleapis.com compute.googleapis.com dns.googleapis.com

To create a Managed Service for Apache Kafka cluster, we can use the gcloud managed-kafka clusters create command by specifying the cluster ID, location, number of vCPUs (cpu), RAM (memory), and subnets.

export CLUSTER_ID=<cluster-id>
export PROJECT_ID=<gcp-project-id>
export PROJECT_NUMBER=<gcp-project-number>
export REGION=<gcp-region>

gcloud managed-kafka clusters create $CLUSTER_ID \
  --location=$REGION \
  --cpu=3 \
  --memory=3GiB \
  --subnets=projects/$PROJECT_ID/regions/$REGION/subnetworks/default \
  --async

Set up a client VM

To connect to the Kafka cluster, Kpow must run on a machine with network access to it. In this setup, we use a Google Cloud Compute Engine virtual machine (VM). The VM must be located in the same region as the Kafka cluster and deployed within the same VPC and subnet specified during the cluster's configuration. We can create the client VM using the command shown below. We also attach the http-server tag to the VM, which allows HTTP traffic and enables browser access to the Kpow instance.

gcloud compute instances create kafka-test-instance \
  --scopes=https://www.googleapis.com/auth/cloud-platform \
  --tags=http-server \
  --subnet=projects/$PROJECT_ID/regions/$REGION/subnetworks/default \
  --zone=$REGION-a

Also, we need to update the permissions of the default service account used by the client VM. To ensure that the Kpow instance running on the VM has full access to Managed Service for Apache Kafka resources, bind the predefined admin role (roles/managedkafka.admin) to the service account. This grants Kpow the necessary administrative privileges. For more fine-grained access control within a Kafka cluster, it is recommended to use Kafka ACLs. The Enterprise Edition of Kpow provides robust support for it - see Kpow's ACL management documentation for more details.

gcloud projects add-iam-policy-binding $PROJECT_ID \
  --member="serviceAccount:$PROJECT_NUMBER-compute@developer.gserviceaccount.com" \
  --role=roles/managedkafka.admin

Launch a Kpow Instance

Once our client VM is up and running, we'll connect to it using the SSH-in-browser tool provided by Google Cloud. After establishing the connection, install Docker Engine, as Kpow will be launched using Docker. Refer to the official installation and post-installation guides for detailed instructions.

With Docker ready, we'll then create Kpow's configuration file (e.g., gcp-trial.env). This file defines Kpow's connection settings to the Google managed kafka cluster and include Kpow license details. To get started, confirm that a valid Kpow license is in place, whether we're using the Community or Enterprise edition.

The main section has the following config variables. The ENVIRONMENT_NAME is a display label used within Kpow to identify the Kafka environment, while the BOOTSTRAP value specifies the Kafka bootstrap server address, which Kpow uses to establish a connection. Connection security is managed through SASL over SSL, as indicated by the SECURITY_PROTOCOL value. The SASL_MECHANISM is set to OAUTHBEARER, enabling OAuth-based authentication. To facilitate this, the SASL_LOGIN_CALLBACK_HANDLER_CLASS is configured to use Google's GcpLoginCallbackHandler, which handles OAuth token management for Kafka authentication. Lastly, SASL_JAAS_CONFIG specifies the JAAS login module used for OAuth-based authentication.

As mentioned, this configuration file also contains our Kpow license details and the path to the Google service account key file. These are essential not only for activating and running Kpow but also for enabling its access to the Kafka cluster.

## Managed Service for Apache Kafka Cluster Configuration
ENVIRONMENT_NAME=GCP Kafka Cluster
BOOTSTRAP=bootstrap.<cluster-id>.<gcp-region>.managedkafka.<gcp-project-id>.cloud.goog:9092
SECURITY_PROTOCOL=SASL_SSL
SASL_MECHANISM=OAUTHBEARER
SASL_LOGIN_CALLBACK_HANDLER_CLASS=com.google.cloud.hosted.kafka.auth.GcpLoginCallbackHandler
SASL_JAAS_CONFIG=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;

## Your License Details
LICENSE_ID=<license-id>
LICENSE_CODE=<license-code>
LICENSEE=<licensee>
LICENSE_EXPIRY=<license-expiry>
LICENSE_SIGNATURE=<license-signature>

Once the gcp-trial.env file is prepared, we'll launch the Kpow instance using the docker run command below. This command maps port 3000 (Kpow's UI port) to port 80 on the host. As a result, we can access the Kpow UI in the browser simply at http://<vm-external-ip>, with no port number needed.

docker run --pull=always -p 80:3000 --name kpow \
  --env-file gcp-trial.env -d factorhouse/kpow-ce:latest

Monitor and Manage Resources

With Kpow now running, we can use its user-friendly UI to monitor brokers, create a topic, send a message to it, and then watch that message get consumed.

Conclusion

By following the steps outlined in this post, we have successfully established a Google Cloud Managed Service for Apache Kafka cluster and deployed a Kpow instance on a Compute Engine VM. With this setup, we can immediately start exploring and managing Kafka brokers and topics, giving us valuable insights into our Kafka environment and streamlining operations.

Kpow is packed with powerful features, and it also integrates seamlessly with Kafka connectors deployed on Google Cloud Managed Kafka Connect clusters. This opens up a world of possibilities for managing data pipelines with ease. Stay tuned as we continue to roll out more integration examples in the future, enabling us all to unlock even more value from our Kafka and Kpow setups.

Overview

Apache Kafka is a cornerstone of modern real-time data pipelines, facilitating high-throughput, low-latency event streaming.

Managing Kafka infrastructure, particularly at scale, presents significant operational challenges. To address this, Amazon Managed Streaming for Apache Kafka (MSK) provides a fully managed service - simplifying the provisioning, configuration, patching, and scaling of Kafka clusters. While MSK handles the infrastructure heavy lifting, effective management and control are still crucial for maintaining cluster health, performance, and reliability.

This article provides a comprehensive walkthrough to setting up an Amazon MSK cluster and integrating it with Kpow for Apache Kafka, a powerful tool for managing and monitoring Kafka environments. It walks through provisioning AWS infrastructure, configuring authentication with the OAUTHBEARER mechanism using AWS IAM, setting up a client EC2 instance within the same VPC, deploying Kpow via Docker, and using Kpow's UI to monitor/manage brokers, topics, and messages.

Whether you manage production Kafka workloads or are evaluating management solutions, this guide provides practical steps for effectively managing and monitoring Kafka clusters on AWS.

About Factor House

Factor House is a leader in real-time data tooling, empowering engineers with innovative solutions for Apache Kafka® and Apache Flink®.

Our flagship product, Kpow for Apache Kafka, is the market-leading enterprise solution for Kafka management and monitoring.

Explore our live multi-cluster demo environment or grab a free Community license and dive into streaming tech on your laptop with Factor House Local.

Set up an EC2 instance

For this post, we're utilizing an Ubuntu-based EC2 instance. Since the MSK cluster will be configured to accept traffic only from within the same VPC, this instance will serve as our primary access point for interacting with the cluster. To ensure connectivity and control, the instance must:

• Be launched in the same VPC as the MSK cluster
• Allow inbound HTTP (port 80) and SSH (port 22) traffic via its security group

We use the AWS Command Line Interface (CLI) to provision and manage AWS resources throughout the demo. If the CLI is not already installed, follow the official AWS CLI user guide for setup and configuration instructions.

As Kpow is designed to manage/monitor Kafka clusters and associated resources, we can give administrative privileges to it. For more fine-grained access control within a Kafka cluster, we can rely on Apache Kafka ACLs, and the Enterprise Edition of Kpow provides robust support for it - see Kpow's ACL management documentation for more details.

Below shows example policies that can be attached.

Option 1: Admin Access to ALL MSK Clusters in the Region/Account

This policy allows listing/describing all clusters and performing any data-plane action (kafka-cluster:*) on any cluster within the specified region and account.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "kafka-cluster:*",
            "Resource": "arn:aws:kafka:<REGION>:<ACCOUNT-ID>:cluster/*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "kafka:ListClusters",
                "kafka:DescribeCluster",
                "kafka:GetBootstrapBrokers"
            ],
            "Resource": "*"
        }
    ]
}

Option 2: Admin Access to a Specific LIST of MSK Clusters

This policy allows listing/describing all clusters but restricts the powerful kafka-cluster:* data-plane actions to only the specific clusters listed in the Resource array.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "kafka-cluster:*",
            "Resource": [
                "arn:aws:kafka:<REGION>:<ACCOUNT-ID>:cluster/<CLUSTER-NAME-1>/<GUID-1>",
                "arn:aws:kafka:<REGION>:<ACCOUNT-ID>:cluster/<CLUSTER-NAME-2>/<GUID-2>"
                // Add more cluster ARNs here as needed following the same pattern
                // "arn:aws:kafka:<REGION>:<ACCOUNT-ID>:cluster/<CLUSTER-NAME-3>/<GUID-3>"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "kafka:ListClusters",
                "kafka:DescribeCluster",
                "kafka:GetBootstrapBrokers"
            ],
            "Resource": "*"
        }
    ]
}

Create a MSK Cluster

While Kpow supports both provisioned and serverless MSK clusters, we'll use an MSK Serverless cluster in this post.

First, create a security group for the Kafka cluster. This security group allows traffic on Kafka port 9098 from:

• Itself (for intra-cluster communication), and
• The EC2 instance's security group (for Kpow access within the same VPC).

VPC_ID=<vpc-ic>
SUBNET_ID1=<subnet-id-1>
SUBNET_ID2=<subnet-id-2>
SUBNET_ID3=<subnet-id-3>
EC2_SG_ID=<ec2-security-group-id>
CLUSTER_NAME=<cluster-name>
REGION=<aws-region>

SG_ID=$(aws ec2 create-security-group \
  --group-name ${CLUSTER_NAME}-sg \
  --description "Security group for $CLUSTER_NAME" \
  --vpc-id "$VPC_ID" \
  --region "$REGION" \
  --query 'GroupId' --output text)

## Allow traffic from itself
aws ec2 authorize-security-group-ingress \
  --group-id "$SG_ID" \
  --protocol tcp \
  --port 9098 \
  --source-group $SG_ID \
  --region "$REGION"

## Allow traffic from EC2 instance
aws ec2 authorize-security-group-ingress \
  --group-id "$SG_ID" \
  --protocol tcp \
  --port 9098 \
  --source-group $EC2_SG_ID \
  --region "$REGION"

Next, create an MSK serverless cluster. We use the aws kafka create-cluster-v2 command with a JSON configuration that specifies:

• VPC subnet and security group,
• SASL/IAM-based client authentication.

read -r -d '' SERVERLESS_JSON <<EOF
{
    "VpcConfigs": [
        {
          "SubnetIds": ["$SUBNET_ID1", "$SUBNET_ID2", "$SUBNET_ID3"],
          "SecurityGroupIds": ["$SG_ID"]
        }
    ],
    "ClientAuthentication": {
      "Sasl": {
        "Iam": { "Enabled": true }
        }
    }
}
EOF

aws kafka create-cluster-v2 \
    --cluster-name "$CLUSTER_NAME" \
    --serverless "$SERVERLESS_JSON" \
    --region "$REGION"

Launch a Kpow Instance

We'll connect to the EC2 instance via SSH and install Docker Engine, as Kpow relies on Docker for its launch. For detailed instructions, please refer to the official installation and post-installation guides.

With Docker ready, we'll create Kpow's configuration file (e.g., aws-trial.env). This file defines Kpow's core settings for connecting to the MSK cluster and includes Kafka connection details, licensing information, and AWS credentials.

The main section defines how Kpow connects to the MSK cluster:

• ENVIRONMENT_NAME: A human-readable name for the Kafka environment shown in the Kpow UI.
• BOOTSTRAP: The Kafka bootstrap server URL for the MSK Serverless cluster (e.g., boot-xxxxxxxx.c2.kafka-serverless.<region>.amazonaws.com:9098).
• KAFKA_VARIANT: Set this to MSK_SERVERLESS to ensure Kpow creates its internal topics with the constrained topic configuration properties and service limitations specific to MSK Serverless.

Secure communication with the cluster is established using SASL over SSL:

• SECURITY_PROTOCOL: Set to SASL_SSL to enable encrypted client-server communication.
• SASL_MECHANISM: Set to AWS_MSK_IAM to use AWS IAM for Kafka client authentication.
• SASL_JAAS_CONFIG: Specifies the use of the IAMLoginModule provided by Amazon for secure authentication.
• SASL_CLIENT_CALLBACK_HANDLER_CLASS: Points to IAMClientCallbackHandler, which automates the process of retrieving and refreshing temporary credentials via IAM.

Finally, the configuration file includes Kpow license details and AWS credentials. These are essential not only to activate and run Kpow but also for it to access the Kafka cluster.

## Managed Service for Apache Kafka Cluster Configuration
ENVIRONMENT_NAME=MSK Serverless
BOOTSTRAP=boot-<cluster-identifier>.c2.kafka-serverless.<aws-region>.amazonaws.com:9098
KAFKA_VARIANT=MSK_SERVERLESS
SECURITY_PROTOCOL=SASL_SSL
SASL_MECHANISM=AWS_MSK_IAM
SASL_JAAS_CONFIG=software.amazon.msk.auth.iam.IAMLoginModule required;
SASL_CLIENT_CALLBACK_HANDLER_CLASS=software.amazon.msk.auth.iam.IAMClientCallbackHandler

## Your License Details
LICENSE_ID=<license-id>
LICENSE_CODE=<license-code>
LICENSEE=<licensee>
LICENSE_EXPIRY=<license-expiry>
LICENSE_SIGNATURE=<license-signature>

## AWS Credentials
AWS_ACCESS_KEY_ID=<aws-access-key>
AWS_SECRET_ACCESS_KEY=<aws-secret-access-key>
AWS_SESSION_TOKEN=<aws-session-token> # Optional
AWS_REGION=<aws-region>

With the aws-trial.env file created, we'll use the following docker run command to launch Kpow. This command forwards Kpow's internal UI port (3000) to port 80 on the host EC2 instance, enabling us to access the Kpow UI in a browser at http://<ec2-public-ip> without specifying a port.

docker run --pull=always -p 80:3000 --name kpow \
  --env-file aws-trial.env -d factorhouse/kpow-ce:latest

Monitor and Manage Resources

With Kpow launched, we can now step through a typical workflow using its user-friendly UI: from monitoring brokers and creating a topic, to sending a message and observing its journey to consumption.

Conclusion

In summary, this guide walked through setting up a fully managed Kafka environment on AWS using Amazon MSK Serverless and Kpow. By leveraging MSK Serverless for Kafka infrastructure and Kpow for observability and control, we can streamline operations while gaining deep insight into our data pipelines. The process included provisioning AWS resources, configuring a secure cluster with IAM-based authentication, and deploying Kpow via Docker Compose with environment-specific and security-conscious settings.

Once connected, Kpow provides an intuitive interface to monitor brokers, manage topics, produce and consume messages, and track consumer lag in real time. Beyond the basics, it offers advanced features like schema inspection, Kafka Connect monitoring, RBAC enforcement, and audit visibility - helping teams shift from reactive troubleshooting to proactive, insight-driven operations. Together, Amazon Managed Streaming for Apache Kafka (MSK) and Kpow form a robust foundation for building and managing high-performance, secure, real-time streaming applications on AWS.

Overview

In Apache Kafka, offset management actions such as clearing, resetting, and skipping play a vital role in controlling consumer group behavior.

• Resetting offsets enables consumers to reprocess messages from a specific point in time or offset - commonly used during recovery, testing, or reprocessing scenarios.
• Skipping offsets allows consumers to move past specific records, such as malformed or corrupted records, without disrupting the entire processing flow.
• Clearing offsets removes committed offsets for a consumer group or member, effectively resetting its consumption history and causing it to re-consume.

Together, these capabilities are essential tools for maintaining data integrity, troubleshooting issues, and ensuring robust and flexible stream processing.

Kpow version 94.2 enhances consumer group management capabilities, providing greater control and visibility into Kafka consumption. This article provides a step-by-step guide on how to manage consumer offsets in Kpow. We will walk through these features using simple Kafka producer and consumer clients.

About Factor House

Factor House is a leader in real-time data tooling, empowering engineers with innovative solutions for Apache Kafka® and Apache Flink®.

Our flagship product, Kpow for Apache Kafka, is the market-leading enterprise solution for Kafka management and monitoring.

Explore our live multi-cluster demo environment or grab a free Community license and dive into streaming tech on your laptop with Factor House Local.

Prerequisites

To manage consumer group offsets in Kpow, the logged-in user must have appropriate permissions, including the GROUP_EDIT action. For more information, refer to the User Authorization section of the Kpow documentation.

Also, if your Kafka cluster has ACLs enabled, the Kafka user specified in Kpow's cluster connection must have the necessary permissions for Kpow to operate correctly. You can find the full list of required permissions in the Minimum ACL Permissions guide.

Development Environment

At Factor House, we recently created a GitHub repository called examples. As the name suggests, it holds Factor House product feature and integration examples.

The Kafka clients used in this post can be found in the offset-management folder. Also, we use Factor House Local to deploy a Kpow instance and a 3-node Kafka cluster.

We can set up a local development environment as follows.

## clone examples
$ git clone https://github.com/factorhouse/examples
$ cd examples

## install python packages in a virtual environment
$ python -m venv venv
$ source venv/bin/activate
$ pip install -r features/offset-management/requirements.txt 

## clone factorhouse-local
$ git clone https://github.com/factorhouse/factorhouse-local

## show key files and folders
$ tree -L 1 factorhouse-local/ features/offset-management/
# factorhouse-local
# ├── LICENSE
# ├── README.md
# ├── compose-analytics.yml
# ├── compose-flex-community.yml
# ├── compose-flex-trial.yml
# ├── compose-kpow-community.yml
# ├── compose-kpow-trial.yml
# ├── compose-pinot.yml
# ├── images
# ├── quickstart
# └── resources
# features/offset-management/
# ├── README.md
# ├── consumer.py
# ├── producer.py
# └── requirements.txt

# 5 directories, 12 files

We can use either the Kpow Community or Enterprise edition. To get started, let's make sure a valid Kpow license is available. For details on how to request and configure a license, refer to this section of the project README.

In this post, we’ll be using the Community edition.

$ docker compose -f factorhouse-local/compose-kpow-community.yml up -d

$ docker compose -f factorhouse-local/compose-kpow-community.yml ps
# NAME              IMAGE                                   COMMAND                  SERVICE     CREATED          STATUS                    PORTS
# connect           confluentinc/cp-kafka-connect:7.8.0     "/etc/confluent/dock…"   connect     44 seconds ago   Up 42 seconds (healthy)   0.0.0.0:8083->8083/tcp, [::]:8083->8083/tcp, 9092/tcp
# kafka-1           confluentinc/cp-kafka:7.8.0             "/etc/confluent/dock…"   kafka-1     44 seconds ago   Up 42 seconds             0.0.0.0:9092->9092/tcp, [::]:9092->9092/tcp
# kafka-2           confluentinc/cp-kafka:7.8.0             "/etc/confluent/dock…"   kafka-2     44 seconds ago   Up 42 seconds             9092/tcp, 0.0.0.0:9093->9093/tcp, [::]:9093->9093/tcp
# kafka-3           confluentinc/cp-kafka:7.8.0             "/etc/confluent/dock…"   kafka-3     44 seconds ago   Up 42 seconds             9092/tcp, 0.0.0.0:9094->9094/tcp, [::]:9094->9094/tcp
# kpow-ce           factorhouse/kpow-ce:latest              "/usr/local/bin/kpow…"   kpow        44 seconds ago   Up 15 seconds             0.0.0.0:3000->3000/tcp, [::]:3000->3000/tcp
# schema_registry   confluentinc/cp-schema-registry:7.8.0   "/etc/confluent/dock…"   schema      44 seconds ago   Up 42 seconds             0.0.0.0:8081->8081/tcp, [::]:8081->8081/tcp
# zookeeper         confluentinc/cp-zookeeper:7.8.0         "/etc/confluent/dock…"   zookeeper   44 seconds ago   Up 43 seconds             2888/tcp, 0.0.0.0:2181->2181/tcp, [::]:2181->2181/tcp, 3888/tcp

Kafka Clients

The Kafka producer interacts with a Kafka cluster using the confluent_kafka library to create a topic and produce messages. It first checks if a specified Kafka topic exists and creates it if necessary - a topic having a single partition is created for simplicity. The app then generates 10 JSON messages containing the current timestamp (both in ISO 8601 and Unix epoch format) and sends them to the Kafka topic, using a callback function to log success or failure upon delivery. Configuration values like the Kafka bootstrap servers and topic name are read from environment variables or default to localhost:9092 and offset-management.

import os
import time
import datetime
import json
import logging

from confluent_kafka import Producer, Message, KafkaError
from confluent_kafka.admin import AdminClient, NewTopic

def topic_exists(admin_client: AdminClient, topic_name: str):
    ## check if a topic exists
    metadata = admin_client.list_topics()
    for value in iter(metadata.topics.values()):
        logging.info(value.topic)
        if value.topic == topic_name:
            return True
    return False

def create_topic(admin_client: AdminClient, topic_name: str):
    ## create a new topic
    new_topic = NewTopic(topic_name, num_partitions=1, replication_factor=1)
    result_dict = admin_client.create_topics([new_topic])
    for topic, future in result_dict.items():
        try:
            future.result()
            logging.info(f"Topic {topic} created")
        except Exception as e:
            logging.info(f"Failed to create topic {topic}: {e}")

def callback(err: KafkaError, event: Message):
    ## callback function that gets triggered when a message is delivered
    if err:
        logging.info(
            f"Produce to topic {event.topic()} failed for event: {event.key()}"
        )
    else:
        value = event.value().decode("utf8")
        logging.info(
            f"Sent: {value} to partition {event.partition()}, offset {event.offset()}."
        )

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)

    BOOTSTRAP_SERVERS = os.getenv("BOOTSTRAP_SERVERS", "localhost:9092")
    TOPIC_NAME = os.getenv("TOPIC_NAME", "offset-management")

    ## create a topic
    conf = {"bootstrap.servers": BOOTSTRAP_SERVERS}
    admin_client = AdminClient(conf)
    if not topic_exists(admin_client, TOPIC_NAME):
        create_topic(admin_client, TOPIC_NAME)

    ## producer messages
    producer = Producer(conf)
    for _ in range(10):
        dt = datetime.datetime.now()
        epoch = int(dt.timestamp() * 1000)
        ts = dt.isoformat(timespec="seconds")
        producer.produce(
            topic=TOPIC_NAME,
            value=json.dumps({"epoch": epoch, "ts": ts}).encode("utf-8"),
            key=str(epoch),
            on_delivery=callback,
        )
        producer.flush()
        time.sleep(1)

The Kafka consumer polls messages from a Kafka topic using the confluent_kafka library. It configures a Kafka consumer with parameters like bootstrap.servers, group.id, and auto.offset.reset. The consumer subscribes to a specified topic, and a callback function (assignment_callback) is used to log when partitions are assigned to the consumer. The main loop continuously polls for messages, processes them by decoding their values, logs the message details (including partition and offset), and commits the message offset. If any errors occur, the script raises a KafkaException. The script gracefully handles a user interrupt (KeyboardInterrupt), ensuring proper cleanup by closing the consumer connection.

import os
import logging
from typing import List

from confluent_kafka import Consumer, KafkaException, TopicPartition, Message


def assignment_callback(_: Consumer, partitions: List[TopicPartition]):
    ## callback function that gets triggered when a topic partion is assigned
    for p in partitions:
        logging.info(f"Assigned to {p.topic}, partiton {p.partition}")


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)

    BOOTSTRAP_SERVERS = os.getenv("BOOTSTRAP_SERVERS", "localhost:9092")
    TOPIC_NAME = os.getenv("TOPIC_NAME", "offset-management")

    conf = {
        "bootstrap.servers": BOOTSTRAP_SERVERS,
        "group.id": f"{TOPIC_NAME}-group",
        "auto.offset.reset": "earliest",
        "enable.auto.commit": False,
    }
    consumer = Consumer(conf)
    consumer.subscribe([TOPIC_NAME], on_assign=assignment_callback)

    try:
        while True:
            message: Message = consumer.poll(1.0)
            if message is None:
                continue
            if message.error():
                raise KafkaException(message.error())
            else:
                value = message.value().decode("utf8")
                partition = message.partition()
                logging.info(
                    f"Reveived {value} from partition {message.partition()}, offset {message.offset()}."
                )
                consumer.commit(message)
    except KeyboardInterrupt:
        logging.warning("cancelled by user")
    finally:
        consumer.close()

Before diving into offset management, let's first create a Kafka topic named offset-management and produce 10 sample messages.

$ python features/offset-management/producer.py 
# ...
# INFO:root:Topic offset-management created
# INFO:root:Sent: {"epoch": 1744847497314, "ts": "2025-04-17T09:51:37"} to partition 0, offset 0.
# INFO:root:Sent: {"epoch": 1744847499328, "ts": "2025-04-17T09:51:39"} to partition 0, offset 1.
# INFO:root:Sent: {"epoch": 1744847500331, "ts": "2025-04-17T09:51:40"} to partition 0, offset 2.
# INFO:root:Sent: {"epoch": 1744847501334, "ts": "2025-04-17T09:51:41"} to partition 0, offset 3.
# INFO:root:Sent: {"epoch": 1744847502337, "ts": "2025-04-17T09:51:42"} to partition 0, offset 4.
# INFO:root:Sent: {"epoch": 1744847503339, "ts": "2025-04-17T09:51:43"} to partition 0, offset 5.
# INFO:root:Sent: {"epoch": 1744847504342, "ts": "2025-04-17T09:51:44"} to partition 0, offset 6.
# INFO:root:Sent: {"epoch": 1744847505344, "ts": "2025-04-17T09:51:45"} to partition 0, offset 7.
# INFO:root:Sent: {"epoch": 1744847506347, "ts": "2025-04-17T09:51:46"} to partition 0, offset 8.
# INFO:root:Sent: {"epoch": 1744847507349, "ts": "2025-04-17T09:51:47"} to partition 0, offset 9.

Next, let the consumer keep polling messages from the topic. By subscribing the topic, it creates a consumer group named offset-management-group.

$ python features/offset-management/consumer.py 
# INFO:root:Assigned to offset-management, partiton 0
# INFO:root:Reveived {"epoch": 1744847497314, "ts": "2025-04-17T09:51:37"} from partition 0, offset 0.
# INFO:root:Reveived {"epoch": 1744847499328, "ts": "2025-04-17T09:51:39"} from partition 0, offset 1.
# INFO:root:Reveived {"epoch": 1744847500331, "ts": "2025-04-17T09:51:40"} from partition 0, offset 2.
# INFO:root:Reveived {"epoch": 1744847501334, "ts": "2025-04-17T09:51:41"} from partition 0, offset 3.
# INFO:root:Reveived {"epoch": 1744847502337, "ts": "2025-04-17T09:51:42"} from partition 0, offset 4.
# INFO:root:Reveived {"epoch": 1744847503339, "ts": "2025-04-17T09:51:43"} from partition 0, offset 5.
# INFO:root:Reveived {"epoch": 1744847504342, "ts": "2025-04-17T09:51:44"} from partition 0, offset 6.
# INFO:root:Reveived {"epoch": 1744847505344, "ts": "2025-04-17T09:51:45"} from partition 0, offset 7.
# INFO:root:Reveived {"epoch": 1744847506347, "ts": "2025-04-17T09:51:46"} from partition 0, offset 8.
# INFO:root:Reveived {"epoch": 1744847507349, "ts": "2025-04-17T09:51:47"} from partition 0, offset 9.

Offset Management

Clear Offsets

This feature removes committed offsets for an entire consumer group, a specific group member, or a group member's partition assignment. When you trigger Clear offsets and confirm the action, it is scheduled and visible under the Mutations tab. The action remains in a scheduled state until its prerequisite is fulfilled. For group offsets, the prerequisite is that the consumer group must be in an EMPTY state - meaning no active members. To meet this condition, you must manually scale down or stop all instances of the consumer group. This requirement exists because offset-related mutations cannot be applied to a running consumer group. When we stop the Kafka consumer, we see the status becomes Success.

Below shows that the consumer polls messages from the earliest offset when it is restarted.

$ python features/offset-management/consumer.py 
# INFO:root:Assigned to offset-management, partiton 0
# INFO:root:Reveived {"epoch": 1744847497314, "ts": "2025-04-17T09:51:37"} from partition 0, offset 0.
# INFO:root:Reveived {"epoch": 1744847499328, "ts": "2025-04-17T09:51:39"} from partition 0, offset 1.
# INFO:root:Reveived {"epoch": 1744847500331, "ts": "2025-04-17T09:51:40"} from partition 0, offset 2.
# INFO:root:Reveived {"epoch": 1744847501334, "ts": "2025-04-17T09:51:41"} from partition 0, offset 3.
# INFO:root:Reveived {"epoch": 1744847502337, "ts": "2025-04-17T09:51:42"} from partition 0, offset 4.
# INFO:root:Reveived {"epoch": 1744847503339, "ts": "2025-04-17T09:51:43"} from partition 0, offset 5.
# INFO:root:Reveived {"epoch": 1744847504342, "ts": "2025-04-17T09:51:44"} from partition 0, offset 6.
# INFO:root:Reveived {"epoch": 1744847505344, "ts": "2025-04-17T09:51:45"} from partition 0, offset 7.
# INFO:root:Reveived {"epoch": 1744847506347, "ts": "2025-04-17T09:51:46"} from partition 0, offset 8.
# INFO:root:Reveived {"epoch": 1744847507349, "ts": "2025-04-17T09:51:47"} from partition 0, offset 9.

Reset Offsets

Kpow provides powerful and flexible controls for managing consumer group offsets across various levels of granularity. You can adjust offsets based on different dimensions, including:

• Whole group assignments - reset offsets for the entire consumer group at once.
• Host-level assignments - target offset changes to consumers running on specific hosts.
• Topic-level assignments - reset offsets for specific topics within the group.
• Partition-level assignments - make fine-grained adjustments at the individual partition level.

In addition to flexible targeting, Kpow supports multiple methods for resetting offsets, tailored to different operational needs:

• Offset - reset to a specific offset value.
• Timestamp - rewind or advance to the earliest offset with a timestamp equal to or later than the given epoch timestamp.
• Datetime (Local) - use a human-readable local date and time to select the corresponding offset.

These options enable precise control over Kafka consumption behavior, whether for incident recovery, message reprocessing, or routine operational adjustments.

In this example, we reset the offset of topic partition 0 using the Timestamp method. After entering the timestamp value 1744847503339, Kpow displays both the current committed offset (Commit Offset) and the calculated target offset (New Offset). Once we confirm the action, it appears in the Mutations tab as a scheduled mutation. Similar to the Clear offsets feature, the mutation remains pending until the consumer is stopped - at which point its status updates to Success. We can then verify the updated offset value directly within the group member's assignment record, confirming that the reset has taken effect.

As expected, the consumer polls messages from the new offset.

$ python features/offset-management/consumer.py 
# INFO:root:Assigned to offset-management, partiton 0
# INFO:root:Reveived {"epoch": 1744847503339, "ts": "2025-04-17T09:51:43"} from partition 0, offset 5.
# INFO:root:Reveived {"epoch": 1744847504342, "ts": "2025-04-17T09:51:44"} from partition 0, offset 6.
# INFO:root:Reveived {"epoch": 1744847505344, "ts": "2025-04-17T09:51:45"} from partition 0, offset 7.
# INFO:root:Reveived {"epoch": 1744847506347, "ts": "2025-04-17T09:51:46"} from partition 0, offset 8.
# INFO:root:Reveived {"epoch": 1744847507349, "ts": "2025-04-17T09:51:47"} from partition 0, offset 9.

Skip Offsets

We can demonstrate the Skip offsets feature in two steps. First, reset the offset of topic partition 0 to 8 and then stop the consumer. This ensures the offset update is applied successfully. Next, trigger the Skip offsets action, which advances the offset from 8 to 9, effectively skipping the message at offset 8. We can verify that the new offset has taken effect by checking the updated value in the group member's assignment record, and it confirms that the change has been applied as expected.

We can also verify the change by restarting the consumer.

$ python features/offset-management/consumer.py 
# INFO:root:Assigned to offset-management, partiton 0
# INFO:root:Reveived {"epoch": 1744847507349, "ts": "2025-04-17T09:51:47"} from partition 0, offset 9.

Conclusion

Managing Kafka consumer offsets is crucial for reliable data processing and operational flexibility. Actions such as clearing, resetting (by offset, timestamp, or datetime), and skipping offsets provide powerful capabilities for debugging, data recovery, and bypassing problematic messages.

This article demonstrated these concepts using simple Kafka clients and showcased how Kpow offers an intuitive interface with granular controls - spanning entire groups down to individual partitions - to execute these essential offset management tasks effectively. Kpow significantly simplifies these complex Kafka operations, ultimately providing developers and operators with enhanced visibility and precise control over their Kafka consumption patterns.

Overview

Managing Apache Kafka within a platform like Confluent Cloud provides significant advantages in scalability and managed services.

This post details the process of integrating Kpow with Confluent Cloud resources such as Kafka brokers, Schema Registry, Kafka Connect, and ksqlDB. Once Kpow is up and running we will explore the powerful user interface to monitor and interact with our Confluent Cloud Kafka cluster.

About Factor House

Factor House is a leader in real-time data tooling, empowering engineers with innovative solutions for Apache Kafka® and Apache Flink®.

Our flagship product, Kpow for Apache Kafka, is the market-leading enterprise solution for Kafka management and monitoring.

Explore our live multi-cluster demo environment or grab a free Community license and dive into streaming tech on your laptop with Factor House Local.

Launch a Kpow Instance

To integrate Kpow with Confluent Cloud, we will first prepare a configuration file (e.g., confluent-trial.env). This file will define Kpow's connection settings to the Confluent Cloud resources and include Kpow license details. Before proceeding, confirm that a valid Kpow license is in place, whether we're using the Community or Enterprise edition.

The primary section of this configuration file details the connection settings Kpow needs to securely communicate with the Confluent Cloud environment, encompassing Kafka, Schema Registry, Kafka Connect, and ksqlDB.

To achieve the objectives outlined in this guide, where we explore enabling Kpow to monitor brokers and topics, as well as to create a topic, produce messages to it, and consume them from a Confluent Cloud cluster, the Kafka Cluster Connection section is mandatory. This section provides the fundamental parameters Kpow requires to establish a secure and authenticated link to the target Confluent Cloud environment. Without this vital configuration, Kpow will be unable to discover brokers, or perform any data operations such as creating topics, producing messages, or consuming them - which are the core functionalities we are setting out to configure in this walkthrough. The specific details required within this section, particularly the BOOTSTRAP server addresses and the API Key/Secret pair used within the SASL_JAAS_CONFIG (acting as username and password for SASL/PLAIN authentication), will be unique to the specific Confluent Cloud environment being configured. Comprehensive instructions on generating these API keys, understanding their associated permissions necessary for Kpow's operations, and locating the bootstrap server information for that environment can be found within the official Confluent Cloud documentation (for API keys) and guides on connecting clients.

Kafka Cluster Connection

• ENVIRONMENT_NAME=Confluent Cloud: A label shown in the Kpow UI to identify this environment.
• BOOTSTRAP=<bootstrap-server-addresses>: The address(es) of the Kafka cluster's bootstrap servers. These are used by Kpow to discover brokers and establish a connection.
• SECURITY_PROTOCOL=SASL_SSL: Specifies that communication with the Kafka cluster uses SASL over SSL for secure authentication and encryption.
• SASL_MECHANISM=PLAIN: Indicates the use of the PLAIN mechanism for SASL authentication.
• SASL_JAAS_CONFIG=...: Contains the username and password used to authenticate with Confluent Cloud using the PLAIN mechanism.
• SSL_ENDPOINT_IDENTIFICATION_ALGORITHM=https: Ensures the broker's identity is verified via hostname verification, as required by Confluent Cloud.

API Key for Enhanced Confluent Features

• CONFLUENT_API_KEY and CONFLUENT_API_SECRET: Used to authenticate with Confluent Cloud's REST APIs for additional metadata or control plane features.
• CONFLUENT_DISK_MODE=COMPLETE: Instructs Kpow to use full disk-based persistence, useful in managed cloud environments where Kpow runs remotely from the Kafka cluster.

Schema Registry Integration

• SCHEMA_REGISTRY_NAME=...: Display name for the Schema Registry in the Kpow UI.
• SCHEMA_REGISTRY_URL=...: The HTTPS endpoint of the Confluent Schema Registry.
• SCHEMA_REGISTRY_AUTH=USER_INFO: Specifies the authentication method (in this case, basic user info).
• SCHEMA_REGISTRY_USER / SCHEMA_REGISTRY_PASSWORD: The credentials to authenticate with the Schema Registry.

Kafka Connect Integration

• CONNECT_REST_URL=...: The URL of the Kafka Connect REST API.
• CONNECT_AUTH=BASIC: Indicates that basic authentication is used to secure the Connect endpoint.
• CONNECT_BASIC_AUTH_USER / CONNECT_BASIC_AUTH_PASS: The credentials for accessing the Kafka Connect REST interface.

ksqlDB Integration

• KSQLDB_NAME=...: A label for the ksqlDB instance as shown in Kpow.
• KSQLDB_HOST and KSQLDB_PORT: Define the location of the ksqlDB server.
• KSQLDB_USE_TLS=true: Enables secure communication with ksqlDB via TLS.
• KSQLDB_BASIC_AUTH_USER / KSQLDB_BASIC_AUTH_PASSWORD: Authentication credentials for ksqlDB.
• KSQLDB_USE_ALPN=true: Enables Application-Layer Protocol Negotiation (ALPN), which is typically required when connecting to Confluent-hosted ksqlDB endpoints over HTTPS.

The configuration file also includes Kpow license details, which, as mentioned, are necessary to activate and run Kpow.

## Confluent Cloud Configuration
ENVIRONMENT_NAME=Confluent Cloud
BOOTSTRAP=<bootstrap-server-addresses>
SECURITY_PROTOCOL=SASL_SSL
SASL_MECHANISM=PLAIN
SASL_JAAS_CONFIG=org.apache.kafka.common.security.plain.PlainLoginModule required username="username" password="password";
SSL_ENDPOINT_IDENTIFICATION_ALGORITHM=https
CONFLUENT_API_KEY=<confluent-api-key>
CONFLUENT_API_SECRET=<confluent-api-secret>
CONFLUENT_DISK_MODE=COMPLETE
SCHEMA_REGISTRY_NAME=Confluent Schema Registry
SCHEMA_REGISTRY_URL=<schema-registry-url>
SCHEMA_REGISTRY_AUTH=USER_INFO
SCHEMA_REGISTRY_USER=<schema-registry-username>
SCHEMA_REGISTRY_PASSWORD=<schema-registry-password>
CONNECT_REST_URL=<connect-rest-url>
CONNECT_AUTH=BASIC
CONNECT_BASIC_AUTH_USER=<connect-basic-auth-username>
CONNECT_BASIC_AUTH_PASS=<connect-basic-auth-password>
KSQLDB_NAME=Confluent ksqlDB
KSQLDB_HOST=<ksqldb-hose>
KSQLDB_PORT=443
KSQLDB_USE_TLS=true
KSQLDB_BASIC_AUTH_USER=<ksqldb-basic-auth-username>
KSQLDB_BASIC_AUTH_PASSWORD=<ksqldb-basic-auth-password>
KSQLDB_USE_ALPN=true

## Your License Details
LICENSE_ID=<license-id>
LICENSE_CODE=<license-code>
LICENSEE=<licensee>
LICENSE_EXPIRY=<license-expiry>
LICENSE_SIGNATURE=<license-signature>

Once the confluent-trial.env file is prepared, we can launch the Kpow instance using the following docker run command:

docker run --pull=always -p 3000:3000 --name kpow \
  --env-file confluent-trial.env -d factorhouse/kpow-ce:latest

Monitor and Manage Resources

Once Kpow is launched, we can explore how it enables us to monitor brokers, create a topic, produce a message to it, and observe that message being consumed, all within its user-friendly UI.

Conclusion

By following these steps, we've successfully launched Kpow with Docker and established its connection to our Confluent Cloud. The key to this was our confluent-trial.env file, where we defined Kpow's license details and the essential connection settings for Kafka, Schema Registry, Kafka Connect, and ksqlDB.

The subsequent demonstration highlighted Kpow's intuitive user interface and its core capabilities, allowing us to effortlessly monitor Kafka brokers, create topics, produce messages, and observe their consumption within our Confluent Cloud cluster. This integrated setup not only simplifies the operational aspects of managing Kafka in the cloud but also empowers users with deep visibility and control, ultimately leading to more robust and efficient event streaming architectures. With Kpow connected to Confluent Cloud, we are now well-equipped to manage and optimize Kafka deployments effectively.