Introduction

IBM Event Streams for IBM Cloud is a high-throughput message bus built with Apache Kafka. It is optimized for event ingestion into IBM Cloud and event stream distribution between your services and applications.

Event Streams provides a REST API to help connect your existing systems to your Event Streams Kafka cluster. Using the API, you can integrate Event Streams with any system that supports RESTful APIs.

You can download the OpenAPI specification compliant descriptors for this REST interface. Alternatively, click the three vertical dots next to the IBM Event Streams Admin REST API title in the top left corner of this app, then click the Download OpenAPI definition link that appears.

Other Event Streams APIs

IBM Event Streams also provides the following APIs:

SDKs

Event Streams is based on the open source Apache Kafka, and as such, all SDKs (client libraries) are provided for the open source community, covering the core runtimes listed. For the most up-to-date information about these libraries, refer to the project homepages.

For more information on what is supported, see support summary for all recommended clients.

SDK language support

This Admin REST API supports the four core languages and the SDKs are available as follows.

Endpoint URLs

Administration API endpoint is the kafka_admin_url property in the service key for the service instance. You can use this command to retrieve the kafka_admin_url property property.

$ ibmcloud resource service-key "${service_instance_key_name}" --output json | jq -r '.[]|.credentials.kafka_admin_url'

In addition, the Content-type header must be set to application/json.

Error handling

This API uses standard HTTP response codes to indicate whether a method completed successfully. A 200 response code indicates success. A 400 response code indicates a failure, and a 500 response code indicates an internal system error.

| HTTP error code | Description | Recovery | |-----------------|-------------|----------| | 200 | OK | Message successfully sent to Event Streams. | | 202 | Request accepted| Kafka accepted the request and processes the request. | | 400 | Bad request | Not a valid request. | | 401 | Unauthorized | The authentication header is not set or provided information is not valid. | | 403 | Not authorized to perform the operation | The API key used is missing a certain role. For more details on what role can perform what operation, see Managing access to your Event Streams resources.| | 404 | Not found | Unable to find the topic with the topic name you specified. | | 408 | Request timeout | The request timed out producing to Event Streams. | | 422 | Semantically invalid request| You have a malformed request. | | 500 | Internal server error | The request failed due to an internal server error. | | 503 | Service unavailable | The request failed due to Event Streams brokers being unavailable. |

Error responses carry a JSON body, such as the following:

{"error_code":50301,"message":"Unknown Kafka Error", "incident_id": "17afe715-0ff5-4c49-9acc-a4204244a331"}

Error codes have the format HHHKK, where HHH is the HTTP status code and KK is the Kafka protocol error.

For end-to-end debugging purposes, the transaction ID of every request is returned in the HTTP header X-Global-Transaction-Id. If the header is set on the request, it is honored. If not, it is generated. In the event of a non-200 error code, the transaction ID is also returned in the JSON error response as incident_id.

Authentication

Use one of the following methods to authenticate:

To authenticate using Basic Auth:
Use the user and api_key properties of the above objects as the username and password. Place these values into the Authorization header of the HTTP request in the form Basic <base64 encoding of username and password joined by a single colon (:)>.
To authenticate using a bearer token:
To obtain your token using the IBM Cloud CLI, first log in to IBM Cloud, then run the following command:
```
ibmcloud iam oauth-tokens
```
Place this token in the Authorization header of the HTTP request in the form Bearer<token>. Both API key and JWT tokens are supported.
To authenticate directly using the api_key:
Place the key directly as the value of the X-Auth-Token HTTP header.

The following list shows the headers required to authenticate with CURL:

-H 'Accept: application/json'
-H 'Content-Type: application/json'
-H 'Authorization: Bearer ${TOKEN}' (use with bearer token)
-H 'X-Auth-Token: ${API_KEY}' (use with API Key)

The following shows an example of how to authenticate with Go:

	// Create Authenticator
	var authenticator core.Authenticator

	if apiKey != "" {
		var err error
		// Create an Basic IAM authenticator.
		authenticator, err = core.NewBasicAuthenticator("token", apiKey)
		if err != nil {
			fmt.Printf("failed to create new basic authenticator: %s\n", err.Error())
			os.Exit(1)
		}
	} else {
		var err error
		// Create an IAM Bearer Token authenticator.
		authenticator, err = core.NewBearerTokenAuthenticator(bearerToken)
		if err != nil {
			fmt.Printf("failed to create new bearer token authenticator: %s\n", err.Error())
			os.Exit(1)
		}
	}
	// End Authenticator

The following example shows how to authenticate with Python:

## Create Authenticator
if not KAFKA_ADMIN_URL:
    print("Please set env KAFKA_ADMIN_URL")
    exit(1)

if not API_KEY and not BEARER_TOKEN:
    print("Please set either an API_KEY or a BEARER_TOKEN")
    exit(1)

if API_KEY and BEARER_TOKEN:
    print("Please set either an API_KEY or a BEARER_TOKEN not both")
    exit(1)

if API_KEY:
    # Create an Basic IAM authenticator.
    authenticator = BasicAuthenticator('token', API_KEY)
else :
    # Create an IAM Bearer Token authenticator.
    authenticator = BasicAuthenticator('token', BEARER_TOKEN)

service = AdminrestV1(
    authenticator = authenticator
    )
## End Authenticator

## Create Service
base_url = KAFKA_ADMIN_URL
service.set_service_url(base_url)
## End Create Service

def list_topics(service):
    # Set up parameter values
    topic_filter = ''
    # Invoke list method.
    try:
        response = service.list_topics(
            topic_filter=topic_filter,
            )

        if response.status_code == HTTPStatus.OK:
            if not response.result :
                print("\tnothing to list")
                return
            for topic in response.result: 
                 print("\t" + topic["name"])
    except:
        print("\tError Listing Topics")
    # func.end

The following example shows how to authenticate with Java:

        // Create Authenticator
        Authenticator authenticator;

        if (apiKey != null && !apiKey.isEmpty()) {
            // Create an Basic IAM authenticator.
            authenticator = new BasicAuthenticator("token", apiKey);
        } else {
            // Create an IAM Bearer Token authenticator.
            authenticator = new BearerTokenAuthenticator(bearerToken);
        }

        // Create Service - Construct the service client.
        Adminrest service = new Adminrest(serviceName, authenticator);
        // End Authenticator

The following example shows how to authenticate with Node:

// Create Authenticator
if (KAFKA_ADMIN_URL === undefined || !KAFKA_ADMIN_URL) {
  console.log('Please set env KAFKA_ADMIN_URL');
  throw new Error('error KAFKA_ADMIN_URL not set');
}

if ((API_KEY === undefined || !API_KEY) && (BEARER_TOKEN === undefined || !BEARER_TOKEN)) {
  console.log('Please set either an API_KEY or a BEARER_TOKEN');
  throw new Error('error: API_KEY or BEARER_TOKEN not set');
}

if (API_KEY && BEARER_TOKEN) {
  console.log('Please set either an API_KEY or a BEARER_TOKEN not both');
  throw new Error('error: API_KEY and BEARER_TOKEN can not both be set');
}

if (API_KEY) {
  // Create an Basic IAM authenticator.
  authenticator = new BasicAuthenticator({
    username: 'token',
    password: API_KEY,
  });
} else {
  // Create an IAM Bearer Token authenticator.
  authenticator = new BearerTokenAuthenticator({
    bearerToken: BEARER_TOKEN,
  });
}
// End Authenticator

Obtaining the Kafka version

You can obtain the Kafka version of the cluster using this API. The path of the API is /admin/info

Run a command like the following:

-$ curl -u token:<redacted API key> https://mh-int-edotest3-cpsrfxk-6fc3ea743571ba7392ba4250a5946fd9-0000.us-south.containers.appdomain.cloud/admin/info

The command provides output like the following:

{"health":{"encryption_key":"Active","kafka":"Healthy","key_protect_key":"Active"},"versions":{"kafka":"3.3"}}

Code setup

The following sections show how to set up the code for each language.

No code setup is required for CURL.

N/A

The following example shows how to set up with Go:

// Code Setup
import (
	"fmt"
	"net/http"
	"os"

	"github.com/IBM/eventstreams-go-sdk/pkg/adminrestv1"
	"github.com/IBM/go-sdk-core/v4/core"
)

// End Code Setup

The following shows an example of how to set up with Python:

## Code Setup
from typing import Set
from ibm_cloud_sdk_core.authenticators import BasicAuthenticator
from eventstreams_sdk.adminrest_v1 import *
import os
from http import HTTPStatus

SERVICE_NAME = 'adminrest_v1'
KAFKA_ADMIN_URL = os.getenv('KAFKA_ADMIN_URL')
BEARER_TOKEN= os.getenv('BEARER_TOKEN')
API_KEY= os.getenv('API_KEY')

## End Code Setup

## Create Authenticator
if not KAFKA_ADMIN_URL:
    print("Please set env KAFKA_ADMIN_URL")
    exit(1)

if not API_KEY and not BEARER_TOKEN:
    print("Please set either an API_KEY or a BEARER_TOKEN")
    exit(1)

if API_KEY and BEARER_TOKEN:
    print("Please set either an API_KEY or a BEARER_TOKEN not both")
    exit(1)

if API_KEY:
    # Create an Basic IAM authenticator.
    authenticator = BasicAuthenticator('token', API_KEY)
else :
    # Create an IAM Bearer Token authenticator.
    authenticator = BasicAuthenticator('token', BEARER_TOKEN)

service = AdminrestV1(
    authenticator = authenticator
    )
## End Authenticator

## Create Service
base_url = KAFKA_ADMIN_URL
service.set_service_url(base_url)
## End Create Service

def list_topics(service):
    # Set up parameter values
    topic_filter = ''
    # Invoke list method.
    try:
        response = service.list_topics(
            topic_filter=topic_filter,
            )

        if response.status_code == HTTPStatus.OK:
            if not response.result :
                print("\tnothing to list")
                return
            for topic in response.result: 
                 print("\t" + topic["name"])
    except:
        print("\tError Listing Topics")
    # func.end

The following shows an example of how to set up with Java:

// Code Setup
package com.ibm.cloud.adminrest.v1;

import java.util.Arrays;
import java.util.List;
import com.ibm.cloud.eventstreams_sdk.adminrest.v1.Adminrest;
import com.ibm.cloud.eventstreams_sdk.adminrest.v1.model.ListTopicsOptions;
import com.ibm.cloud.eventstreams_sdk.adminrest.v1.model.TopicDetail;
import com.ibm.cloud.eventstreams_sdk.adminrest.v1.model.CreateTopicOptions;
import com.ibm.cloud.eventstreams_sdk.adminrest.v1.model.DeleteTopicOptions;
import com.ibm.cloud.eventstreams_sdk.adminrest.v1.model.UpdateTopicOptions;
import com.ibm.cloud.eventstreams_sdk.adminrest.v1.model.ReplicaAssignment;
import com.ibm.cloud.eventstreams_sdk.adminrest.v1.model.GetMirroringActiveTopicsOptions;
import com.ibm.cloud.eventstreams_sdk.adminrest.v1.model.MirroringActiveTopics;
import com.ibm.cloud.eventstreams_sdk.adminrest.v1.model.ReplaceMirroringTopicSelectionOptions;
import com.ibm.cloud.eventstreams_sdk.adminrest.v1.model.MirroringTopicSelection;
import com.ibm.cloud.eventstreams_sdk.adminrest.v1.model.GetMirroringTopicSelectionOptions;
import com.ibm.cloud.eventstreams_sdk.adminrest.v1.model.ListConsumerGroupsOptions;
import com.ibm.cloud.eventstreams_sdk.adminrest.v1.model.GetConsumerGroupOptions;
import com.ibm.cloud.eventstreams_sdk.adminrest.v1.model.GroupDetail;
import com.ibm.cloud.eventstreams_sdk.adminrest.v1.model.UpdateConsumerGroupOptions;
import com.ibm.cloud.eventstreams_sdk.adminrest.v1.model.GroupResetResult;
import com.ibm.cloud.eventstreams_sdk.adminrest.v1.model.DeleteConsumerGroupOptions;
import com.ibm.cloud.sdk.core.security.Authenticator;
import com.ibm.cloud.sdk.core.security.BasicAuthenticator;
import com.ibm.cloud.sdk.core.security.BearerTokenAuthenticator;
import com.ibm.cloud.sdk.core.http.Response;
import com.ibm.cloud.sdk.core.http.HttpStatus;

public class AdminrestExample {

    private AdminrestExample() {
    }
// End Code Setup

The following example shows how to set up with Node:

// Code Setup
const HTTP = require('http');
const util = require('util');
const KAFKA_ADMIN_URL = process.env.KAFKA_ADMIN_URL;
const API_KEY = process.env.API_KEY;
const BEARER_TOKEN = process.env.BEARER_TOKEN;
const NewAdminrestV1 = require('../dist/adminrest/v1');
const { BasicAuthenticator } = require('../dist/auth');
const { BearerTokenAuthenticator } = require('../dist/auth');
const { NoAuthAuthenticator } = require('../dist/auth');

const topicName = 'test-topic';
let authenticator = new NoAuthAuthenticator({});
/* eslint-disable no-console */

// Code Setup End

// Create Authenticator
if (KAFKA_ADMIN_URL === undefined || !KAFKA_ADMIN_URL) {
  console.log('Please set env KAFKA_ADMIN_URL');
  throw new Error('error KAFKA_ADMIN_URL not set');
}

if ((API_KEY === undefined || !API_KEY) && (BEARER_TOKEN === undefined || !BEARER_TOKEN)) {
  console.log('Please set either an API_KEY or a BEARER_TOKEN');
  throw new Error('error: API_KEY or BEARER_TOKEN not set');
}

if (API_KEY && BEARER_TOKEN) {
  console.log('Please set either an API_KEY or a BEARER_TOKEN not both');
  throw new Error('error: API_KEY and BEARER_TOKEN can not both be set');
}

if (API_KEY) {
  // Create an Basic IAM authenticator.
  authenticator = new BasicAuthenticator({
    username: 'token',
    password: API_KEY,
  });
} else {
  // Create an IAM Bearer Token authenticator.
  authenticator = new BearerTokenAuthenticator({
    bearerToken: BEARER_TOKEN,
  });
}
// End Authenticator

Create service

The following sections show how to create a service for each language.

No service setup is required for CURL.

N/A

The following example shows how to create a service with Go:

	// Create Service
	serviceAPI, serviceErr := adminrestv1.NewAdminrestV1(&adminrestv1.AdminrestV1Options{
		URL:           URL,
		Authenticator: authenticator,
	})
	// End Create Service

The following example shows how to create a service with Python:

## Create Service
base_url = KAFKA_ADMIN_URL
service.set_service_url(base_url)
## End Create Service

def list_topics(service):
    # Set up parameter values
    topic_filter = ''
    # Invoke list method.
    try:
        response = service.list_topics(
            topic_filter=topic_filter,
            )

        if response.status_code == HTTPStatus.OK:
            if not response.result :
                print("\tnothing to list")
                return
            for topic in response.result: 
                 print("\t" + topic["name"])
    except:
        print("\tError Listing Topics")
    # func.end

The following example shows how to create a service with Java:

        // Create Service - Construct the service client.
        Adminrest service = new Adminrest(serviceName, authenticator);
        // End Authenticator

The following example shows how to create a service with Node:

// Create Service
// Construct the service client.
const adminrest = new NewAdminrestV1({
  authenticator,
  serviceUrl: KAFKA_ADMIN_URL,
});
// End Create Service

Event Streams Admin REST API | IBM Cloud API Docs