Apache Kafka：使用java方式操作消费组和重置分区偏移量（admin api）

Posted 2021-09-06 你是小KS

当前版本：kafka_2.12-2.8.0

1. 声明

当前内容主要为本人学习和测试使用java方式操作消费组和重置分区偏移量,主要参考：Apache Kafka官方文档

主要为：

1. 使用java方式实现增删查消费组
2. 使用admin api方式重置分区偏移量

2. 准备条件

1. 使用命令行方式创建一个带分区的复制系数的topic：test

./bin/kafka-topics.sh --create --topic test --replication-factor 1 --partitions 1 --bootstrap-server 192.168.1.101:9092

2. 查看创建后的topic

./bin/kafka-topics.sh --describe --topic test --bootstrap-server 192.168.1.101:9092

maven依赖

<dependency>
	<groupId>org.apache.kafka</groupId>
	<artifactId>kafka_2.12</artifactId>
	<version>2.8.0</version>
</dependency>
<dependency>
	<groupId>org.apache.kafka</groupId>
	<artifactId>kafka-clients</artifactId>
	<version>2.8.0</version>
</dependency>

3. 使用java方式增删查消费组

消费组的创建(完全是由创建消费者连接当前的kafka的group.id来创建消费组)，相同的消费组共同一个topic中的消息，消费消息后kafka记录偏移量

消费者操作：

public static void main(String[] args) throws Exception {
		Properties props = new Properties();// 创建配置

		// 设置bootstrap.servers并指定当前的kafka的启动的目标地址
		props.put(KafkaProperties.bootstrapServersName, "192.168.1.101:9092");
		props.put("group.id", "testGroup");
		props.put("enable.auto.commit", "true");
		props.put("auto.commit.interval.ms", "1000");
		props.put("session.timeout.ms", "30000");
		// 设置当前的key的解序列化所使用的类
		props.put(KafkaProperties.keyDeserializer, KafkaProperties.stringDeserializerName);
		// 设置当前的value的解序列化所使用的类
		props.put(KafkaProperties.valueDeserializer, KafkaProperties.stringDeserializerName);

		// 创建Kafka的消费者，专门用于消费Kafka中的特定的topic中的消息
		@SuppressWarnings("resource")
		KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

		// Kafka Consumer subscribes list of topics here.
		// 采用订阅的方式消费当前主题的消息
		consumer.subscribe(Arrays.asList("test"));
		// print the topic name
		System.out.println("Subscribed to topic test" );
		while (true) {
			// 循环的方式拉取数据,其中等待时间为100毫秒
			ConsumerRecords<String, String> records = consumer.poll(100);
			for (ConsumerRecord<String, String> record : records) {
				System.out.println(record);
			}
			

		}

	}

开启后立刻就关闭，此时这个消费组就会自动创建！

admin api操作：

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
		Properties props = new Properties();
		props.put("bootstrap.servers", "192.168.1.101:9092");
		/* props.put("client.id", "consumer-testGroup-1"); */ //这里不能有，否则报错
		AdminClient client = KafkaAdminClient.create(props);
		listConsumerGroup(client);
		listConsumerGroupOffset(client, "testGroup");
		
		// 删除已脱机的消费组
		deleteConsumerGroup(client, "testGroup");

		client.close(2l, TimeUnit.SECONDS);
	}

	/**
	 * 
	 * @author hy
	 * @createTime 2021-08-14 09:23:36
	 * @description 列出当前的消费组
	 * @param client
	 *
	 */
	private static void listConsumerGroup(AdminClient client) {
		ListConsumerGroupsResult listConsumerGroups = client.listConsumerGroups();
		try {
			Collection<ConsumerGroupListing> collection = listConsumerGroups.all().get();
			System.out.println(collection);
		} catch (InterruptedException | ExecutionException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	/**
	 * 
	 * @author hy
	 * @createTime 2021-08-14 09:27:27
	 * @description 通过当前的组id方式获取当前的偏移量
	 * @param client
	 * @param groupId
	 *
	 */
	private static void listConsumerGroupOffset(AdminClient client, String groupId) {
		ListConsumerGroupOffsetsResult listConsumerGroupOffsets = client.listConsumerGroupOffsets(groupId);
		try {
			Map<TopicPartition, OffsetAndMetadata> map = listConsumerGroupOffsets.partitionsToOffsetAndMetadata().get();
			System.out.println(map);
		} catch (InterruptedException | ExecutionException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	/**
	 * 
	 * @author hy
	 * @createTime 2021-08-14 09:30:47
	 * @description 查看消费组详情
	 * @param client
	 * @param groupId
	 *
	 */
	private static void describeConsumerGroups(AdminClient client, String groupId) {
		DescribeConsumerGroupsResult describeConsumerGroups = client.describeConsumerGroups(Arrays.asList(groupId));
		try {
			Map<String, ConsumerGroupDescription> map = describeConsumerGroups.all().get();
			System.out.println(map.get(groupId));
		} catch (InterruptedException | ExecutionException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}


	/**
	 * 
	 * @author hy
	 * @createTime 2021-08-14 09:35:59
	 * @description 删除当前的消费者组
	 * @param client
	 * @param groupId
	 *
	 */
	private static void deleteConsumerGroup(AdminClient client, String groupId) {
		DeleteConsumerGroupsResult deleteConsumerGroups = client.deleteConsumerGroups(Arrays.asList(groupId));
		try {
			Void void1 = deleteConsumerGroups.all().get();
			// 只要不报错就表示删除正常
			System.out.println(void1);
		} catch (InterruptedException | ExecutionException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

执行之后，必须当消费组中没有任何消费者存在的时候才可以删除,否则会报错的！错误为：The group is not empty.

命令行方式

## 查看所有的消费组
./kafka-consumer-groups.sh --bootstrap-server 192.168.1.101:9092 --list
## 查看某个消费组的详细信息
./bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group
## 查看某个消费组的成员
./kafka-consumer-groups.sh --bootstrap-server 192.168.1.101:9092 --describe --group my-group --members
## 删除消费组
./kafka-consumer-groups.sh --bootstrap-server 192.168.1.101:9092  --delete --group my-group --group my-group2

4.使用java方式重置偏移量

注意：重置偏移量的前提！(本人测试后)

1. 所有的消费者都已经关闭(断开kafka进行消费消费)
2. 所有的生产者都已经关闭(不再连接kafka进行生产消息)
3. 重置的分区偏移量时候当前topic必须为XXX-数值的分区(这个一定要存在)
4. admin的api不要带有client.id的属性

上面任何一个存在问题,在重置分区偏移量的时候就会出现一下错误：java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.UnknownMemberIdException: Failed altering consumer group offsets for the following partitions:[XXXX]

下面是重置偏移量的代码

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
		Properties props = new Properties();
		props.put("bootstrap.servers", "192.168.1.101:9092");
		/* props.put("client.id", "consumer-testGroup-1"); */ //这里不能有，否则报错
		AdminClient client = KafkaAdminClient.create(props);
		listConsumerGroup(client);
		listConsumerGroupOffset(client, "testGroup");
		//describeConsumerGroups(client, "console-consumer-35180");
		resetConsumerGroupOffset(client, "test",0,"testGroup");
		client.close(2l, TimeUnit.SECONDS);
	}


	/**
	 * 
	 * @author hy
	 * @createTime 2021-08-14 09:38:25
	 * @description 
	 * @param client
	 * @param groupId
	 *
	 */
	private static void resetConsumerGroupOffset(AdminClient client, String topicName,int partition,String groupId) {
		Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<TopicPartition, OffsetAndMetadata>();
		offsets.put(new TopicPartition(topicName,partition), new OffsetAndMetadata(0));
		AlterConsumerGroupOffsetsResult alterConsumerGroupOffsets = client.alterConsumerGroupOffsets(groupId, offsets);
		
		// 执行报错
		// java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.UnknownMemberIdException: Failed altering consumer group offsets for the following partitions: 
		try {
			Void void1 = alterConsumerGroupOffsets.all().get();
			System.out.println(void1);
		} catch (InterruptedException | ExecutionException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

测试
首先使用生产者向test这个topic进行生产消息，并连续发送，最后开启前面的消费者进行消费，此时偏移量达到最大，然后关闭消费者，最后打开消费者发现不再消费(zookeeper记录了偏移量)，使用重置topic的分区偏移量后，再次打开消费者，此时发现消费者继续从0开始消费了

1.第一次打开消费者

2.关闭消费者，再次打开消费者

3.再次关闭消费者，打开重置偏移量(这个地方可能会由于关闭消费者的太快，出现报错，等一下就行了)

这个是重置成功的结果(不报错)

4.再次打开消费者

再次消费成功！

在测试中最容易出现的错误就是：org.apache.kafka.common.errors.UnknownMemberIdException: Failed altering consumer group offsets for the following partitions: [test-0],注意消费者关闭后等待时间和其他问题，最后的结果是测试成功