Brute_Move_Log_Dir.java

package xyz.util.kafka;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AlterReplicaLogDirsResult;
import org.apache.kafka.common.TopicPartitionReplica;

public class Brute_Move_Log_Dir {

    public static void main(String[] args) {
        Properties prop = new Properties();

        prop.setProperty("bootstrap.servers", "kafka-dev01.domain.local:9092");
        String topic_name = "testk2kcl10";
        Integer partition = 10;
        Integer broker_id = 1002;
        String move_replica_to_dir = "/kafka-logs2";

        // build admin client
        AdminClient admin = AdminClient.create(prop);

        //move replica from where to where
        Map<TopicPartitionReplica, String> replicaAssignment = new HashMap<TopicPartitionReplica, String>();
        TopicPartitionReplica tpr = new TopicPartitionReplica(topic_name, partition, broker_id); //topic, partition, brokerid
        replicaAssignment.put(tpr, move_replica_to_dir);

        //execute the command
        AlterReplicaLogDirsResult result = admin.alterReplicaLogDirs(replicaAssignment);
        try {
            result.all().get();
            System.out.println(result.all());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

    }

}