private static Map<TopicPartition, List<Long>> parseJsonData(int version, JsonValue js) throws JsonMappingException {
        if (version == 1) {
            JsonValue partitions = js.asJsonObject().get("partitions")
                .orElseThrow(() -> new AdminOperationException("Missing partitions field"));

            Map<TopicPartition, List<Long>> res = new HashMap<>();

            Iterator<JsonValue> iterator = partitions.asJsonArray().iterator();

            while (iterator.hasNext()) {
                JsonObject partitionJs = iterator.next().asJsonObject();

                String topic = partitionJs.apply("topic").to(STRING);
                int partition = partitionJs.apply("partition").to(INT);
                long offset = partitionJs.apply("offset").to(LONG);

                res.computeIfAbsent(new TopicPartition(topic, partition), k -> new ArrayList<>()).add(offset);
            }

            return res;
        }

        throw new AdminOperationException("Not supported version field value " + version);
    }
A command for delete records of the given partitions down to the specified offset.
tools/src/main/java/org/apache/kafka/tools/DeleteRecordsCommand.java