Apache Kafka:使用java方式操作消费组和重置分区偏移量(admin api)

Posted 你是小KS

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Apache Kafka:使用java方式操作消费组和重置分区偏移量(admin api)相关的知识,希望对你有一定的参考价值。

当前版本: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
  1. 查看创建后的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],注意消费者关闭后等待时间和其他问题,最后的结果是测试成功

以上是关于Apache Kafka:使用java方式操作消费组和重置分区偏移量(admin api)的主要内容,如果未能解决你的问题,请参考以下文章

kafka集群配置和java编写生产者消费者操作例子

Apache Kafka:使用java方式操作stream(实现官方的wordcount)

Apache Kafka:使用java和命令行方式操作topic(admin api)

kafka消费者java版本读取不到消息怎么办

从Apache Kafka到Apache Spark安全地读取数据

是否有任何 Apache Kafka 消费者滞后检查器?