Obtain the consumer offset of a consumer group

Prerequisites

• Simple Log Service is activated. For more information, see Activate Simple Log Service.

• The Simple Log Service SDK for Python is initialized. For more information, see Initialize Simple Log Service SDK for Python.

Parameter description

def get_check_point(self, project, logstore, consumer_group, shard=-1):

Request parameters

Response parameters

For information about the response parameters, see GetCheckPoint.

Sample code

from aliyun.log import LogClient
import os


def main():
    # In this example, the AccessKey ID and AccessKey secret are obtained from environment variables.
    access_key_id = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID', '')
    access_key_secret = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET', '')
    # Specify a Simple Log Service endpoint.
    endpoint = "cn-hangzhou.log.aliyuncs.com"

    client = LogClient(endpoint, access_key_id, access_key_secret)

    project = "ali-test-peoject"
    logstore = "test-logstore"
    consumer_group = "consumer_group_test"
    shard = -1

    try:
        response = client.get_check_point(project, logstore, consumer_group, shard)
        response.log_print()
    except Exception as e:
        print(f"An error occurred while creating the consumer group: {e}")


if __name__ == '__main__':
    main()

Sample response

ListConsumerGroupCheckPoints:
headers: {'Server': 'AliyunSLS', 'Content-Type': 'application/json', 'Content-Length': '237', 'Connection': 'keep-alive', 'Access-Control-Allow-Origin': '*', 'Date': 'Mon, 25 Nov 2024 10:09:42 GMT', 'x-log-time': '1732529382', 'x-log-requestid': '67444CE605099546E07A82E8'}
count: 2
consumer_group_check_points: [{'shard': 0, 'checkpoint': 'MTczMjUxNTg0ODM1OTg3******==', 'updateTime': 1732519816521348, 'consumer': 'consumer-group-1-B'}, {'shard': 1, 'checkpoint': 'MTczMjUxNTg0ODM2MzEx******==', 'updateTime': 1732519816330366, 'consumer': 'consumer-group-1-A'}]

Process finished with exit code 0

References

• For more information about log consumption APIs, see the following topics:

• For more sample code, see Alibaba Cloud Simple Log Service SDK for Python on GitHub.