To reset offsets of a consumer group, "--reset-offsets" option can be used. This option supports one consumer group at the time. It requires defining following scopes: --all-topics or --topic. One scope must be selected, unless you use '--from-file' scenario. Also, first make sure that the consumer instances are inactive. See KIP-122 for more details. It has 3 execution options: (default) to display which offsets to reset.
--execute : to execute --reset-offsets process.
--export : to export the results to a CSV format.
--reset-offsets also has following scenarios to choose from (at least one scenario must be selected): --to-datetime : Reset offsets to offsets from datetime. Format: 'YYYY-MM-DDTHH:mm:SS.sss'
--to-earliest : Reset offsets to earliest offset.
--to-latest : Reset offsets to latest offset.
--shift-by : Reset offsets shifting current offset by 'n', where 'n' can be positive or negative.
--from-file : Reset offsets to values defined in CSV file.
--to-current : Resets offsets to current offset.
--by-duration : Reset offsets to offset by duration from current timestamp. Format: 'PnDTnHnMnS'
--to-offset : Reset offsets to a specific offset.
Please note, that out of range offsets will be adjusted to available offset end. For example, if offset end is at 10 and offset shift request is of 15, then, offset at 10 will actually be selected. For example, to reset offsets of a consumer group to the latest offset: > bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group consumergroup1 --topic topic1 --to-latest TOPIC PARTITION NEW-OFFSET topic1 0 0 If you are using the old high-level consumer and storing the group metadata in ZooKeeper (i.e. offsets.storage=zookeeper), pass --zookeeper instead of --bootstrap-server: > bin/kafka-consumer-groups.sh --zookeeper localhost:2181 --list Expanding your cluster Adding servers to a Kafka cluster is easy, just assign them a unique broker id and start up Kafka on your new servers. However these new servers will not automatically be assigned any data partitions, so unless partitions are moved to them they won't be doing any work until new topics are created. So usually when you add machines to your cluster you will want to migrate some existing data to these machines. The process of migrating data is manually initiated but fully automated. Under the covers what happens is that Kafka will add the new server as a follower of the partition it is migrating and allow it to fully replicate the existing data in that partition. When the new server has fully replicated the contents of this partition and joined the in-sync replica one of the existing replicas will delete their partition's data. The partition reassignment tool can be used to move partitions across brokers. An ideal partition distribution would ensure even data load and partition sizes across all brokers. The partition reassignment tool does not have the capability to automatically study the data distribution in a Kafka cluster and move partitions around to attain an even load distribution. As such, the admin has to figure out which topics or partitions should be moved around. The partition reassignment tool can run in 3 mutually exclusive modes: --generate: In this mode, given a list of topics and a list of brokers, the tool generates a candidate reassignment to move all partitions of the specified topics to the new brokers. This option merely provides a convenient way to generate a partition reassignment plan given a list of topics and target brokers.
--execute: In this mode, the tool kicks off the reassignment of partitions based on the user provided reassignment plan. (using the --reassignment-json-file option). This can either be a custom reassignment plan hand crafted by the admin or provided by using the --generate option
--verify: In this mode, the tool verifies the status of the reassignment for all partitions listed during the last --execute. The status can be either of successfully completed, failed or in progress
Automatically migrating data to new machines The partition reassignment tool can be used to move some topics off of the current set of brokers to the newly added brokers. This is typically useful while expanding an existing cluster since it is easier to move entire topics to the new set of brokers, than moving one partition at a time. When used to do this, the user should provide a list of topics that should be moved to the new set of brokers and a target list of new brokers. The tool then evenly distributes all partitions for the given list of topics across the new set of brokers. During this move, the replication factor of the topic is kept constant. Effectively the replicas for all partitions for the input list of topics are moved from the old set of brokers to the newly added brokers. For instance, the following example will move all partitions for topics foo1,foo2 to the new set of brokers 5,6. At the end of this move, all partitions for topics foo1 and foo2 will only exist on brokers 5,6. Since the tool accepts the input list of topics as a json file, you first need to identify the topics you want to move and create the json file as follows: > cat topics-to-move.json "topics": ["topic": "foo1", "topic": "foo2"], "version":1 Once the json file is ready, use the partition reassignment tool to generate a candidate assignment: > bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --topics-to-move-json-file topics-to-move.json --broker-list "5,6" --generate Current partition replica assignment "version":1, "partitions":["topic":"foo1","partition":0,"replicas":[2,1], "topic":"foo1","partition":1,"replicas":[1,3], "topic":"foo1","partition":2,"replicas":[3,4], "topic":"foo2","partition":0,"replicas":[4,2], "topic":"foo2","partition":1,"replicas":[2,1], "topic":"foo2","partition":2,"replicas":[1,3]] Proposed partition reassignment configuration "version":1, "partitions":["topic":"foo1","partition":0,"replicas":[6,5], "topic":"foo1","partition":1,"replicas":[5,6], "topic":"foo1","partition":2,"replicas":[6,5], "topic":"foo2","partition":0,"replicas":[5,6], "topic":"foo2","partition":1,"replicas":[6,5], "topic":"foo2","partition":2,"replicas":[5,6]] The tool generates a candidate assignment that will move all partitions from topics foo1,foo2 to brokers 5,6. Note, however, that at this point, the partition movement has not started, it merely tells you the current assignment and the proposed new assignment. The current assignment should be saved in case you want to rollback to it. The new assignment should be saved in a json file (e.g. expand-cluster-reassignment.json) to be input to the tool with the --execute option as follows: > bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file expand-cluster-reassignment.json --execute Current partition replica assignment "version":1, "partitions":["topic":"foo1","partition":0,"replicas":[2,1], "topic":"foo1","partition":1,"replicas":[1,3], "topic":"foo1","partition":2,"replicas":[3,4], "topic":"foo2","partition":0,"replicas":[4,2], "topic":"foo2","partition":1,"replicas":[2,1], "topic":"foo2","partition":2,"replicas":[1,3]] Save this to use as the --reassignment-json-file option during rollback Successfully started partition reassignments for foo1-0,foo1-1,foo1-2,foo2-0,foo2-1,foo2-2 Finally, the --verify option can be used with the tool to check the status of the partition reassignment. Note that the same expand-cluster-reassignment.json (used with the --execute option) should be used with the --verify option: > bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file expand-cluster-reassignment.json --verify Status of partition reassignment: Reassignment of partition [foo1,0] is completed Reassignment of partition [foo1,1] is still in progress Reassignment of partition [foo1,2] is still in progress Reassignment of partition [foo2,0] is completed Reassignment of partition [foo2,1] is completed Reassignment of partition [foo2,2] is completed Custom partition assignment and migration The partition reassignment tool can also be used to selectively move replicas of a partition to a specific set of brokers. When used in this manner, it is assumed that the user knows the reassignment plan and does not require the tool to generate a candidate reassignment, effectively skipping the --generate step and moving straight to the --execute step For instance, the following example moves partition 0 of topic foo1 to brokers 5,6 and partition 1 of topic foo2 to brokers 2,3: The first step is to hand craft the custom reassignment plan in a json file: > cat custom-reassignment.json "version":1,"partitions":["topic":"foo1","partition":0,"replicas":[5,6],"topic":"foo2","partition":1,"replicas":[2,3]] Then, use the json file with the --execute option to start the reassignment process: > bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file custom-reassignment.json --execute Current partition replica assignment "version":1, "partitions":["topic":"foo1","partition":0,"replicas":[1,2], "topic":"foo2","partition":1,"replicas":[3,4]] Save this to use as the --reassignment-json-file option during rollback Successfully started partition reassignments for foo1-0,foo2-1 The --verify option can be used with the tool to check the status of the partition reassignment. Note that the same custom-reassignment.json (used with the --execute option) should be used with the --verify option: > bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file custom-reassignment.json --verify Status of partition reassignment: Reassignment of partition [foo1,0] is completed Reassignment of partition [foo2,1] is completed Decommissioning brokers The partition reassignment tool does not have the ability to automatically generate a reassignment plan for decommissioning brokers yet. As such, the admin has to come up with a reassignment plan to move the replica for all partitions hosted on the broker to be decommissioned, to the rest of the brokers. This can be relatively tedious as the reassignment needs to ensure that all the replicas are not moved from the decommissioned broker to only one other broker. To make this process effortless, we plan to add tooling support for decommissioning brokers in the future. Increasing replication factor Increasing the replication factor of an existing partition is easy. Just specify the extra replicas in the custom reassignment json file and use it with the --execute option to increase the replication factor of the specified partitions. For instance, the following example increases the replication factor of partition 0 of topic foo from 1 to 3. Before increasing the replication factor, the partition's only replica existed on broker 5. As part of increasing the replication factor, we will add more replicas on brokers 6 and 7. The first step is to hand craft the custom reassignment plan in a json file: > cat increase-replication-factor.json "version":1, "partitions":["topic":"foo","partition":0,"replicas":[5,6,7]] Then, use the json file with the --execute option to start the reassignment process: > bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file increase-replication-factor.json --execute Current partition replica assignment "version":1, "partitions":["topic":"foo","partition":0,"replicas":[5]] Save this to use as the --reassignment-json-file option during rollback Successfully started partition reassignment for foo-0 The --verify option can be used with the tool to check the status of the partition reassignment. Note that the same increase-replication-factor.json (used with the --execute option) should be used with the --verify option: > bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file increase-replication-factor.json --verify Status of partition reassignment: Reassignment of partition [foo,0] is completed You can also verify the increase in replication factor with the kafka-topics tool: > bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic foo --describe Topic:fooPartitionCount:1ReplicationFactor:3Configs: Topic: fooPartition: 0Leader: 5Replicas: 5,6,7Isr: 5,6,7 Limiting Bandwidth Usage during Data Migration Kafka lets you apply a throttle to replication traffic, setting an upper bound on the bandwidth used to move replicas from machine to machine. This is useful when rebalancing a cluster, bootstrapping a new broker or adding or removing brokers, as it limits the impact these data-intensive operations will have on users.
Camera Timestamp v3.55 Apk [Patched] [Latest]
2ff7e9595c
Comments