// Module included in the following assemblies:
//
// configuring/assembly-reassign-tool.adoc

[id='proc-generating-reassignment-json-files-{context}']
= Generating a reassignment JSON file to reassign partitions

[role="_abstract"]
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
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
    listeners:
      # ...
      - name: tls
        port: 9093
        type: internal
        tls: true <1>
        authentication:
          type: tls <2>
    # ...
----
<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`
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaTopicApiVersion}
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`
[source,yaml,subs="attributes+"]
----
apiVersion: {KafkaUserApiVersion}
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication: <1>
    type: tls
  authorization:
    type: simple <2>
    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.
+
[source,shell,subs="+quotes"]
kubectl get secret _<cluster_name>_-cluster-ca-cert -o jsonpath='{.data.ca\.p12}' | base64 -d > ca.p12
+
[source,shell,subs="+quotes"]
kubectl 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 Strimzi Kafka image to connect to a running Kafka broker.
+
[source,shell,subs="+quotes,attributes"]
----
kubectl run --restart=Never --image={DockerKafkaImageCurrent} _<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.
+
[source,shell,subs=+quotes]
kubectl 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.
+
[source,shell,subs=+quotes]
kubectl get secret _<kafka_user>_ -o jsonpath='{.data.user\.p12}' | base64 -d > user.p12
+
[source,shell,subs=+quotes]
kubectl 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.
+
[source,shell,subs=+quotes]
kubectl 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.
+
[source,properties,subs="+quotes,attributes"]
----
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.
+
[source,shell,subs=+quotes]
kubectl 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`
[source,json]
----
{
  "version": 1,
  "topics": [
    { "topic": "my-topic"}
  ]
}
----

You can also use this file to xref:proc-changing-topic-replicas-{context}[change the replication factor of a topic]. 
--

. Copy the `_topics.json_` file to the interactive pod container.
+
[source,shell,subs=+quotes]
kubectl cp topics.json _<interactive_pod_name>_:/tmp/topics.json

. Start a shell process in the interactive pod container.
+
[source,shell,subs=+quotes]
kubectl exec -n _<namespace>_ -ti _<interactive_pod_name>_ /bin/bash
+
Replace _<namespace>_ with the Kubernetes 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
[source,shell,subs=+quotes]
----
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
----

[role="_additional-resources"]
.Additional resources

* xref:con-config-kafka-{context}[Configuring Kafka]
* xref:proc-configuring-kafka-topic-{context}[]
* xref:proc-configuring-kafka-user-{context}[]
