Kafka快速入门

123456789101112131415

# 解压jdk压缩包至/opttar zxvf jdk-8u181-linux-x64.tar.gz# 配置jdk环境变量vim /etc/profile# 添加如下配置export JAVA_HOME=/opt/jdk1.8.0_181 export JRE_HOME=$JAVA_HOME/jreexport PATH=$PATH:$JAVA_HOME/bin export CLASSPATH=.:$JAVA_HOME/lib:$JRE_HOME/lib# 使配置生效source /etc/profilejava -version

1234567891011121314

# 解压安装包至/opttar zxvf zookeeper-3.4.12.tar.gz# 配置环境变量vim /etc/profile# 添加如下配置export ZOOKEEPER_HOME=/opt/zookeeper-3.4.12export PATH=$PATH:$ZOOKEEPER_HOME/bin# 使配置生效source /etc/profile# 修改zk配置文件cd $ZOOKEEPER_HOME/confcp zoo_sample.cfg zoo.cfg

12345678910111213

# ZooKeeper 服务器心跳时间， 单位为 mstickTime=2000# 投票选举新 leader 的初始化时间initLimit=10# leader 与 follower 心跳检测最大容忍时间，响应超过 syncLimit*tickTime，leader 认为# follower “死掉 ，从服务器列表中删除 followersyncLimit=5# 数据目录dataDir=/tmp/zookeeper/data# 日志目录dataLogDir=/tmp/zookeeper/log# ZooKeeper 对外服务端口clientPort=2181

12

mkdir -p /tmp/zookeeper/datamkdir -p /tmp/zookeeper/log

1

echo 1 > /tmp/zookeeper/data/myid

12

zkServer.sh startzkServer.sh status

123

server.0=192.168.0.2:2888:3888server.1=192.168.0.3:2888:3888server.2=192.168.0.4:2888:3888

1234567

tar zxvf kafka_2.11-2.0.0.tgz# 配置环境变量vim /etc/profile# 添加如下配置export KAFKA_HOME=/opt/kafka_2.11-2.0.0# 使配置生效source /etc/profile

12345678

# broker 的编号，如果集群中有多个broker，则每个broker的编号需要设置的不同broker.id=0# broker对外提供的服务入口地址listeners=PLAINTEXT://localhost:9092# 存放消息日志文件的地址log.dirs=/tmp/kafka-logs# Kafka所需的ZooKeeper集群地址zookeeper.connect=localhost:2181/kafka

123456

# 直接启动bin/kafka-server-start.sh config/server.properties# 后台启动bin/kafka-server-start.sh -daemon config/server.properties# 或者bin/kafka-server-start.sh config/server.properties &

1

bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic topic-demo --replication-factor 3 --partitions 4

1

bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic topic-demo

1

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topic-demo

1

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic-demo --from-beginning

12345

<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>2.0.0</version></dependency>

1234567891011

public static final String brokerList = "192.168.0.100:9092";public static final String topic = "topic-demo"; public static Properties initConfig() { Properties properties = new Properties(); properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList); properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); properties.put(ProducerConfig.CLIENT_ID_CONFIG, "producer.client.id.demo"); return properties;}

1234567891011121314

public static void main(String[] args) { Properties properties = initConfig(); KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties); ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, "Hello,Kafka!"); try { Future<RecordMetadata> future = producer.send(record); //同步模式，这里会阻塞 RecordMetadata metadata = future.get(); System.out.println(metadata.topic() + "-" + metadata.partition() + ":" + metadata.offset()); } catch (Exception e) { e.printStackTrace(); } producer.close();}

123456789

producer.send(record, new Callback() { public void onCompletion(RecordMetadata metadata, Exception e) { if (e != null) { e.printStackTrace(); } else { System.out.println(metadata.topic() + "-" + metadata.partition() + ":" + metadata.offset()); } }});

12345678

@Data@NoArgsConstructor@AllArgsConstructor@Builderpublic class Company { private String name; private String address;}

1234567891011121314151617181920212223242526272829303132333435

public class CompanySerializer implements Serializer<Company> { public void configure(Map<String, ?> map, boolean b) { } public byte[] serialize(String s, Company data) { if (data == null) { return null; } byte[] name, address; try { if (data.getName() != null) { name = data.getName().getBytes("UTF-8"); } else { name = new byte[0]; } if (data.getAddress() != null) { address = data.getAddress().getBytes("UTF-8"); } else { address = new byte[0]; } ByteBuffer buffer = ByteBuffer.allocate(4 + 4 + name.length + address.length); buffer.putInt(name.length); buffer.put(name); buffer.putInt(address.length); buffer.put(address); return buffer.array(); } catch (UnsupportedEncodingException e) { e.printStackTrace(); } return new byte[0]; } public void close() { }}

1

properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, CompanySerializer.class.getName());

12

Company company = Company.builder().name("myCompany").address("China").build();ProducerRecord<String, Company> record = new ProducerRecord<String, Company>(topic, company);

12345678910111213141516171819

public class DemoPartitioner implements Partitioner { private final AtomicInteger counter = new AtomicInteger(0); public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { List<PartitionInfo> partitionInfos = cluster.partitionsForTopic(topic); int numPartitions = partitionInfos.size(); if (keyBytes == null) { return counter.getAndIncrement() % numPartitions; } else { return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions; } } public void close() { } public void configure(Map<String, ?> map) { }}

1

properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, DemoPartitioner.class.getName());

123456789101112131415161718192021222324

public class ProducerInterceptorPrefix implements ProducerInterceptor<String, String> { private volatile long sendSuccess = 0; private volatile long sendFailure = 0; public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) { String modifiedValue = "prefix1-" + record.value(); return new ProducerRecord<String, String>(record.topic(), record.partition(), record.timestamp(), record.key(), modifiedValue, record.headers()); } public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) { if (e == null) sendSuccess++; else sendFailure++; } public void close() { double successRatio = (double) sendSuccess / (sendSuccess + sendFailure); System.out.println("发送成功率=" + String.format("%f", successRatio % 100) + "%"); } public void configure(Map<String, ?> map) { }}

1

properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, ProducerInterceptor.class.getName());

参数名

默认值

说明

12345678910111213

public static final String brokerList = "192.168.32.128:9092";public static final String topic = "topic-demo";public static final String groupId = "group.demo";public static Properties initConfig() { Properties properties = new Properties(); properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList); properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); properties.put(ConsumerConfig.CLIENT_ID_CONFIG, "consumer.client.id.demo"); properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); return properties;}

1234567891011121314151617181920

public static final AtomicBoolean isRunning = new AtomicBoolean(true);public static void main(String[] args) { Properties props = initConfig(); KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props); consumer.subscribe(Arrays.asList(topic)); try { while (isRunning.get()) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000)); for (ConsumerRecord<String, String> record : records) { System.out.println("topic=" + record.topic() + ", partition=" + record.partition() + ", offset=" + record.offset()); System.out.println("key=" + record.key() + ", value=" + record.value()); } } } catch (Exception e) { e.printStackTrace(); } finally { consumer.close(); }}

1

consumer.subscribe(Arrays.asList(topic));

1

consumer.subscribe(Pattern.compile("topic.*"));

1

consumer.assign(Arrays.asList(new TopicPartition(topic, 0)));

1

List<PartitionInfo> partitionsFor(String topic);

123

consumer.unsubscribe();consumer.subscribe(new ArrayList<String>());consumer.assign(new ArrayList<TopicPartition>());

1234567891011121314151617181920212223242526272829303132333435

public class CompanyDeserializer implements Deserializer<Company> { @Override public void configure(Map<String, ?> configs, boolean isKey) { } @Override public Company deserialize(String topic, byte[] data) { if (data == null) { return null; } if (data.length < 8) { throw new SerializationException("Size of data received is shorter than expected!"); } ByteBuffer buffer = ByteBuffer.wrap(data); int nameLen, addressLen; String name, address; nameLen = buffer.getInt(); byte[] nameBytes = new byte[nameLen]; buffer.get(nameBytes); addressLen = buffer.getInt(); byte[] addressBytes = new byte[addressLen]; buffer.get(addressBytes); try { name = new String(nameBytes, "UTF-8"); address = new String(addressBytes, "UTF-8"); } catch (UnsupportedEncodingException e) { throw new SerializationException("Error occur when deserialize!"); } return new Company(name, address); } @Override public void close() { }}

1

properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, CompanyDeserializer.class.getName());

12345678910

<dependency> <groupId>io.protostuff</groupId> <artifactId>protostuff-core</artifactId> <version>1.5.4</version></dependency><dependency> <groupId>io.protostuff</groupId> <artifactId>protostuff-runtime</artifactId> <version>1.5.4</version></dependency>

1234567891011121314151617

@Overridepublic byte[] serialize(String s, Company data) { if (data == null) { return null; } Schema schema = RuntimeSchema.getSchema(data.getClass()); LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE); byte[] protostuff = null; try { protostuff = ProtostuffIOUtil.toByteArray(data, schema, buffer); } catch (Exception e) { throw new IllegalStateException(e.getMessage(), e); } finally { buffer.clear(); } return protostuff;}

12345678910

@Overridepublic Company deserialize(String topic, byte[] data) { if (data == null) { return null; } Schema<Company> schema = RuntimeSchema.getSchema(Company.class); Company company = new Company(); ProtostuffIOUtil.mergeFrom(data, company, schema); return company;}

123456

ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000)); for (TopicPartition tp : records.partitions()) { for (ConsumerRecord<String, String> record : records.records(tp)) { System.out.println(record.partition() + " : " + record.value()); }}

12345

for (String topic : Arrays.asList("topic1", "topic2")) { for (ConsumerRecord<String, String> record : records.records(topic)) { System.out.println(record.topic() + " : " + record.value()); }}

12

enable.auto.commit=trueauto.commit.interval.ms=5000

1

properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);

123456789101112

final int minBatchSize = 200;List<ConsumerRecord> buffer = new ArrayList<>();while (isRunning.get()) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000)); for (ConsumerRecord<String, String> record : records) { buffer.add(record); } if (buffer.size() >= minBatchSize) { consumer.commitSync(); buffer.clear(); }}

123

long offset = record.offset();TopicPartition partition = new TopicPartition(record.topic(), record.partition());consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(offset + 1)));

12345678910

ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));for (TopicPartition tp : records.partitions()) { List<ConsumerRecord<String, String>> partitionRecords = records.records(tp); for (ConsumerRecord<String, String> record : partitionRecords) { System.out.println(record.partition() + " : " + record.value()); } long lastConsumedOffset = partitionRecords.get(partitionRecords.size() - 1).offset(); consumer.commitSync(Collections.singletonMap(tp, new OffsetAndMetadata(lastConsumedOffset + 1))); }}

1234567891011

consumer.commitAsync(new OffsetCommitCallback() { @Override public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) { if(exception==null){ System.out.println(offsets); } else{ exception.printStackTrace(); } }});

12

public void pause(Collections<TopicPartition> partitions)public void resume(Collections<TopicPartition> partitions)

1

public Set<TopicPartition> paused()

123456789101112131415161718

try { while (isRunning.get()) { consumer.poll(); ... } } catch (WakeupException e){ //忽略错误,退出循环}catch (Exception e) { e.printStackTrace();} finally { //这里可提交消费位移 consumer.close();}...consumer.wakeup();

12345678910111213

consumer.subscribe(Arrays.asList(topic)); Set<TopicPartition> assignment = new HashSet<>();while (assignment.size() == 0) { //如果不为0，说明成功分配到了分区 consumer.poll(Duration.ofMillis(100)); assignment = consumer.assignment(); //获取分配到的分区}for (TopicPartition tp : assignment) { consumer.seek(tp, 10); //指定拉取位移}while (isRunning.get()) { //再次拉取数据 ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000)); ...}

1234

Map<TopicPartition, Long> offsets = consumer.endOffsets(assignment);for (TopicPartition tp : assignment) { consumer.seek(tp, offsets.get(tp)); //这里返回的offset是将要写入最新消息的位置，不需要再加1了}

12

public void seekToBeginning(Collection<TopicPartition> partitions)public void seekToEnd(Collection<TopicPartition> partitions)

1234567891011

Map<TopicPartition, Long> timestampToSearch = new HashMap<>();for (TopicPartition tp : assignment) { timestampToSearch.put(tp, System.currentTimeMillis() - 24 * 3600 * 1000); //查询一天前的消息}Map<TopicPartition, OffsetAndTimestamp> offsets = consumer.offsetsForTimes(timestampToSearch);for (TopicPartition tp : assignment) { OffsetAndTimestamp offsetAndTimestamp = offsets.get(tp); if (offsetAndTimestamp != null) { consumer.seek(tp, offsetAndTimestamp.offset()); }}

12345678910

consumer.subscribe(Arrays.asList(topic), new ConsumerRebalanceListener() { @Override public void onPartitionsRevoked(Collection<TopicPartition> partitions){ //将位移存入数据库 } @Override public void onPartitionsAssigned(Collection<TopicPartition> partitions) { //从数据库中读取位移 }});

1234567891011121314151617181920212223242526272829303132333435

public class ConsumerInterceptorTTL implements ConsumerInterceptor<String, String> { private static final long EXPIRE_INTERVAL = 10 * 1000; @Override public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) { long now = System.currentTimeMillis(); Map<TopicPartition, List<ConsumerRecord<String, String>>> newRecords = new HashMap<>(); for (TopicPartition tp : records.partitions()) { List<ConsumerRecord<String, String>> tpRecords = records.records(tp); List<ConsumerRecord<String, String>> newTpRecords = new ArrayList<>(); for (ConsumerRecord<String, String> record : tpRecords) { if (now - record.timestamp() < EXPIRE_INTERVAL) { newTpRecords.add(record); } } if (!newTpRecords.isEmpty()) { newRecords.put(tp, newTpRecords); } } return new ConsumerRecords<>(newRecords); } @Override public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) { offsets.forEach((tp, offset) -> System.out.println(tp + ":" + offset.offset())); } @Override public void close() { } @Override public void configure(Map<String, ?> configs) { }}

1234567891011121314151617181920212223242526272829303132

public static void main(String[] args) { Properties properties = initConfig(); int consumerThreadNum = 4; for (int i = 0; i < consumerThreadNum; i++) { new KafkaConsumerThread(properties, topic).start(); }}public class KafkaConsumerThread extends Thread { private KafkaConsumer<String, String> kafkaConsumer; public KafkaConsumerThread(Properties properties, String topic) { this.kafkaConsumer = new KafkaConsumer<>(properties); this.kafkaConsumer.subscribe(Arrays.asList(topic)); } @Override public void run() { try { while (true) { ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100)); for (ConsumerRecord<String, String> record : records) { //处理消息 } } } catch (Exception e) { e.printStackTrace(); } finally { kafkaConsumer.close(); } }}

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980

properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);public static void main(String[] args) { Properties properties = initConfig(); KafkaConsumerThread consumerThread = new KafkaConsumerThread(properties, topic, Runtime.getRuntime().availableProcessors()); consumerThread.start();}public class KafkaConsumerThread extends Thread { private KafkaConsumer<String, String> kafkaConsumer; private ExecutorService executorService; private int threadNumber; private Map<TopicPartition, OffsetAndMetadata> offsets; public KafkaConsumerThread(Properties properties, String topic, int threadNumber) { kafkaConsumer = new KafkaConsumer<>(properties); kafkaConsumer.subscribe(Collections.singletonList(topic)); offsets = new HashMap<>(); this.threadNumber = threadNumber; executorService = new ThreadPoolExecutor(threadNumber, threadNumber, 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(1000), new ThreadPoolExecutor.CallerRunsPolicy()); } @Override public void run() { try { while (true) { ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100)); if (!records.isEmpty()) { executorService.submit(new RecordsHandler(records, offsets)); //手动提交位移 synchronized (offsets) { if (!offsets.isEmpty()) { kafkaConsumer.commitSync(offsets); offsets.clear(); } } } } } catch (Exception e) { e.printStackTrace(); } finally { kafkaConsumer.close(); } }}public class RecordsHandler extends Thread { public final ConsumerRecords<String, String> records; private Map<TopicPartition, OffsetAndMetadata> offsets; public RecordsHandler(ConsumerRecords<String, String> records, Map<TopicPartition, OffsetAndMetadata> offsets) { this.records = records; this.offsets = offsets; } @Override public void run() { for (TopicPartition tp : records.partitions()) { List<ConsumerRecord<String, String>> tpRecords = records.records(tp); for (ConsumerRecord<String, String> record : records) { //处理消息 } long lastConsumedOffset = tpRecords.get(tpRecords.size() - 1).offset(); //每个线程记录自己的提交位移 synchronized (offsets) { if (!offsets.containsKey(tp)) { offsets.put(tp, new OffsetAndMetadata(lastConsumedOffset + 1)); } else { long position = offsets.get(tp).offset(); if (position < lastConsumedOffset + 1) { offsets.put(tp, new OffsetAndMetadata(lastConsumedOffset + 1)); } } } } }}

参数名

默认值

说明

1

bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic topic-create --replication-factor 2 --partitions 4

1

bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic topic-create-same --replica-assignment 2:0,0:1,1:2,2:1

1

bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic topic-config --replication-factor 1 --partitions 1 --config cleanup.policy=compact --config max.message.bytes=10000

1

bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic topic-create,topic-config

1

bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topics-with-overrides

1

bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic topic-create --under-replicated-partitions

1

bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic topic-create --unavailable-partitions

1

bin/kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic topic-config --partitions 3

1

bin/kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic topic-config --config segment.bytes=1045877

1

bin/kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic topic-config --delete-config segment.bytes --delete-config cleanup.policy

1

bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic topic-delete

参数名

说明

1

bin/kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type topics --entity-name topic-config

1

bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --entity-type topics --entity-name topic-config --add-config cleanup.policy=compact,max.message.bytes=10000

1

bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --entity-type topics --entity-name topic-config --delete-config cleanup.policy,max.message.bytes

配置类型

entity-type取值

entity-name取值

主题端参数

对应的broker端参数

说明

12345678910111213141516171819202122

Properties properties = new Properties();properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);properties.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);AdminClient client = AdminClient.create(properties);//自动指定分区副本分配方案//NewTopic newTopic = new NewTopic(topic, 4, (short) 1);//或者手动指定分区副本分配方案Map<Integer, List<Integer>> replicasAssignments = new HashMap<>();replicasAssignments.put(0, Arrays.asList(0));replicasAssignments.put(1, Arrays.asList(0));replicasAssignments.put(2, Arrays.asList(0));replicasAssignments.put(3, Arrays.asList(0));NewTopic newTopic = new NewTopic(topic, replicasAssignCreateTopicsResult result = client.createTopics(Collections.singleton(newTopic));try { result.all().get();} catch (InterruptedException e) { e.printStackTrace();} catch (ExecutionException e) { e.printStackTrace();}client.close();

1234

NewTopic newTopic = new NewTopic(topic, replicasAssignments);Map<String, String> configs = new HashMap<>();configs.put("cleanup.policy", "compact");newTopic.configs(configs);

12345678910

ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, topic);DescribeConfigsResult result = client.describeConfigs(Collections.singleton(resource));try { Config config = result.all().get().get(resource); System.out.println(config);} catch (InterruptedException e) { e.printStackTrace();} catch (ExecutionException e) { e.printStackTrace();}

12345678910111213

ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, topic);ConfigEntry entry = new ConfigEntry("cleanup.policy", "compact");Config config = new Config(Collections.singleton(entry));Map<ConfigResource, Config> configs = new HashMap<>();configs.put(resource, config);AlterConfigsResult result = client.alterConfigs(configs);try { result.all().get();} catch (InterruptedException e) { e.printStackTrace();} catch (ExecutionException e) { e.printStackTrace();}

1234567891011

NewPartitions newPartitions = NewPartitions.increaseTo(5);Map<String, NewPartitions> newPartitionsMap = new HashMap<>();newPartitionsMap.put(topic, newPartitions);CreatePartitionsResult result = client.createPartitions(newPartitionsMap);try { result.all().get();} catch (InterruptedException e) { e.printStackTrace();} catch (ExecutionException e) { e.printStackTrace();}

123456789101112131415161718192021

public class PolicyDemo implements CreateTopicPolicy { @Override public void validate(RequestMetadata requestMetadata) throws PolicyViolationException { if (requestMetadata.numPartitions() != null || requestMetadata.replicationFactor() != null) { if (requestMetadata.numPartitions() < 5) { throw new PolicyViolationException("Topic should have at least 5 partitions, received: " + requestMetadata.numPartitions()); } if (requestMetadata.replicationFactor() <= 1) { throw new PolicyViolationException("Topic should have at least 2 replication factor, received: " + requestMetadata.replicationFactor()); } } } @Override public void close() throws Exception { } @Override public void configure(Map<String, ?> configs) { }}

1

bin/kafka-leader-election.sh --bootstrap-server localhost:9092 --election-type preferred --path-to-json-file election.json

1234567891011

{ "partitions": [{ "topic": "foo", "partition": 1 }, { "topic": "foobar", "partition": 2 } ]}

1

bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --generate --topics-to-move-json-file reassign.json --broker-list 0,2

123456

{ "topics": [{ "topic": "topic-reassign" }], "version": 1}

1

bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --execute --reassignment-json-file project.json

1

bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --vertify --reassignment-json-file project.json

1

bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type brokers --entity-name 1 --alter --add-config follower.replication.throttled.rate=1024,leader.replication.throttled.rate=1024

1

bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name topic-throttle --alter --add-config leader.replication.throttled.replicas=[0:0,1:1,2:2],follower.replication.throttled.replicas=[0:1,1:2,2:0]

123456789

{ "partitions": [{ "topic": "topic-throttle", "partition": 0, "replicas": [2, 1, 0], "log_dirs": ["any", "any", "any"] }], "version": 1}