主页
产品
Streams for Apache Kafka
2.7
Deploying and Managing Streams for Apache Kafka on OpenShift
20.2. Generating a reassignment JSON file to reassign partitions

20.2. Generating a reassignment JSON file to reassign partitions

Generate a reassignment JSON file with the kafka-reassign-partitions.sh tool to reassign partitions after scaling a Kafka cluster. Adding or removing brokers does not automatically redistribute the existing partitions. To balance the partition distribution and take full advantage of the new brokers, you can reassign the partitions using the kafka-reassign-partitions.sh tool.

You run the tool from an interactive pod container connected to the Kafka cluster.

The following procedure describes a secure reassignment process that uses mTLS. You’ll need a Kafka cluster that uses TLS encryption and mTLS authentication.

You’ll need the following to establish a connection:

The cluster CA certificate and password generated by the Cluster Operator when the Kafka cluster is created
The user CA certificate and password generated by the User Operator when a user is created for client access to the Kafka cluster

In this procedure, the CA certificates and corresponding passwords are extracted from the cluster and user secrets that contain them in PKCS #12 (.p12 and .password) format. The passwords allow access to the .p12 stores that contain the certificates. You use the .p12 stores to specify a truststore and keystore to authenticate connection to the Kafka cluster.

Prerequisites

You have a running Cluster Operator.

You have a running Kafka cluster based on a Kafka resource configured with internal TLS encryption and mTLS authentication.

Kafka configuration with TLS encryption and mTLS authentication

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
    listeners:
      # ...
      - name: tls
        port: 9093
        type: internal
        tls: true


        authentication:
          type: tls


    # ...

1: Enables TLS encryption for the internal listener.
2: Listener authentication mechanism specified as mutual tls.

The running Kafka cluster contains a set of topics and partitions to reassign.

Example topic configuration for my-topic

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
  name: my-topic
  labels:
    strimzi.io/cluster: my-cluster
spec:
  partitions: 10
  replicas: 3
  config:
    retention.ms: 7200000
    segment.bytes: 1073741824
    # ...

You have a KafkaUser configured with ACL rules that specify permission to produce and consume topics from the Kafka brokers.

Example Kafka user configuration with ACL rules to allow operations on my-topic and my-cluster

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication:


    type: tls
  authorization:
    type: simple


    acls:
      # access to the topic
      - resource:
          type: topic
          name: my-topic
        operations:
          - Create
          - Describe
          - Read
          - AlterConfigs
        host: "*"
      # access to the cluster
      - resource:
          type: cluster
        operations:
          - Alter
          - AlterConfigs
        host: "*"
      # ...
  # ...

1: User authentication mechanism defined as mutual tls.
2: Simple authorization and accompanying list of ACL rules.

Procedure

Extract the cluster CA certificate and password from the <cluster_name>-cluster-ca-cert secret of the Kafka cluster.
```
oc get secret <cluster_name>-cluster-ca-cert -o jsonpath='{.data.ca\.p12}' | base64 -d > ca.p12
```
```
oc get secret <cluster_name>-cluster-ca-cert -o jsonpath='{.data.ca\.password}' | base64 -d > ca.password
```
Replace <cluster_name> with the name of the Kafka cluster. When you deploy Kafka using the Kafka resource, a secret with the cluster CA certificate is created with the Kafka cluster name (<cluster_name>-cluster-ca-cert). For example, my-cluster-cluster-ca-cert.
Run a new interactive pod container using the Streams for Apache Kafka image to connect to a running Kafka broker.
```
oc run --restart=Never --image=registry.redhat.io/amq-streams/kafka-37-rhel9:2.7.0 <interactive_pod_name> -- /bin/sh -c "sleep 3600"
```
Replace <interactive_pod_name> with the name of the pod.
Copy the cluster CA certificate to the interactive pod container.
```
oc cp ca.p12 <interactive_pod_name>:/tmp
```
Extract the user CA certificate and password from the secret of the Kafka user that has permission to access the Kafka brokers.
```
oc get secret <kafka_user> -o jsonpath='{.data.user\.p12}' | base64 -d > user.p12
```
```
oc get secret <kafka_user> -o jsonpath='{.data.user\.password}' | base64 -d > user.password
```
Replace <kafka_user> with the name of the Kafka user. When you create a Kafka user using the KafkaUser resource, a secret with the user CA certificate is created with the Kafka user name. For example, my-user.
Copy the user CA certificate to the interactive pod container.
```
oc cp user.p12 <interactive_pod_name>:/tmp
```
The CA certificates allow the interactive pod container to connect to the Kafka broker using TLS.
Create a config.properties file to specify the truststore and keystore used to authenticate connection to the Kafka cluster.
Use the certificates and passwords you extracted in the previous steps.
```
bootstrap.servers=<kafka_cluster_name>-kafka-bootstrap:9093 
```
1
```
security.protocol=SSL 
```
2
```
ssl.truststore.location=/tmp/ca.p12 
```
3
```
ssl.truststore.password=<truststore_password> 
```
4
```
ssl.keystore.location=/tmp/user.p12 
```
5
```
ssl.keystore.password=<keystore_password> 
```
6
1
The bootstrap server address to connect to the Kafka cluster. Use your own Kafka cluster name to replace <kafka_cluster_name>.
2
The security protocol option when using TLS for encryption.
3
The truststore location contains the public key certificate (ca.p12) for the Kafka cluster.
4
The password (ca.password) for accessing the truststore.
5
The keystore location contains the public key certificate (user.p12) for the Kafka user.
6
The password (user.password) for accessing the keystore.

Copy the config.properties file to the interactive pod container.

oc cp config.properties <interactive_pod_name>:/tmp/config.properties

Prepare a JSON file named topics.json that specifies the topics to move.
Specify topic names as a comma-separated list.
Example JSON file to reassign all the partitions of my-topic
```
{
  "version": 1,
  "topics": [
    { "topic": "my-topic"}
  ]
}
```
You can also use this file to change the replication factor of a topic.

Copy the topics.json file to the interactive pod container.

oc cp topics.json <interactive_pod_name>:/tmp/topics.json

Start a shell process in the interactive pod container.
```
oc exec -n <namespace> -ti <interactive_pod_name> /bin/bash
```
Replace <namespace> with the OpenShift namespace where the pod is running.

Use the kafka-reassign-partitions.sh command to generate the reassignment JSON.

Example command to move the partitions of my-topic to specified brokers

bin/kafka-reassign-partitions.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 \
  --command-config /tmp/config.properties \
  --topics-to-move-json-file /tmp/topics.json \
  --broker-list 0,1,2,3,4 \
  --generate

20.2. Generating a reassignment JSON file to reassign partitions

学习

尝试、购买和销售

社区

關於紅帽

让开源更具包容性

关于红帽文档

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links