kafka PositionSerde 源码

kafka PositionSerde 代码

文件路径：/streams/src/main/java/org/apache/kafka/streams/state/internals/PositionSerde.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.kafka.streams.state.internals;

import org.apache.kafka.streams.query.Position;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

public final class PositionSerde {

    // uninstantiable utility class
    public PositionSerde() {
    }

    public static Position deserialize(final ByteBuffer buffer) {
        final byte version = buffer.get();

        switch (version) {
            case (byte) 0: // actual position, v0
                final int nTopics = buffer.getInt();
                final Map<String, Map<Integer, Long>> position = new HashMap<>(nTopics);
                for (int i = 0; i < nTopics; i++) {
                    final int topicNameLength = buffer.getInt();
                    final byte[] topicNameBytes = new byte[topicNameLength];
                    buffer.get(topicNameBytes);
                    final String topic = new String(topicNameBytes, StandardCharsets.UTF_8);

                    final int numPairs = buffer.getInt();
                    final Map<Integer, Long> partitionOffsets = new HashMap<>(numPairs);
                    for (int j = 0; j < numPairs; j++) {
                        partitionOffsets.put(buffer.getInt(), buffer.getLong());
                    }
                    position.put(topic, partitionOffsets);
                }
                return Position.fromMap(position);
            default:
                throw new IllegalArgumentException(
                        "Unknown version " + version + " when deserializing Position"
                );
        }
    }

    public static ByteBuffer serialize(final Position position) {
        final byte version = (byte) 0;

        int arraySize = Byte.SIZE; // version

        final int nTopics = position.getTopics().size();
        arraySize += Integer.SIZE;

        final ArrayList<String> entries =
                new ArrayList<>(position.getTopics());
        final byte[][] topics = new byte[entries.size()][];

        for (int i = 0; i < nTopics; i++) {
            final String topic = entries.get(i);
            final byte[] topicBytes = topic.getBytes(StandardCharsets.UTF_8);
            topics[i] = topicBytes;
            arraySize += Integer.SIZE; // topic name length
            arraySize += topicBytes.length; // topic name itself

            final Map<Integer, Long> partitionOffsets = position.getPartitionPositions(topic);
            arraySize += Integer.SIZE; // Number of PartitionOffset pairs
            arraySize += (Integer.SIZE + Long.SIZE)
                    * partitionOffsets.size(); // partitionOffsets themselves
        }

        final ByteBuffer buffer = ByteBuffer.allocate(arraySize);
        buffer.put(version);

        buffer.putInt(nTopics);
        for (int i = 0; i < nTopics; i++) {
            buffer.putInt(topics[i].length);
            buffer.put(topics[i]);

            final String topic = entries.get(i);
            final Map<Integer, Long> partitionOffsets = position.getPartitionPositions(topic);
            buffer.putInt(partitionOffsets.size());
            for (final Entry<Integer, Long> partitionOffset : partitionOffsets.entrySet()) {
                buffer.putInt(partitionOffset.getKey());
                buffer.putLong(partitionOffset.getValue());
            }
        }

        buffer.flip();
        return buffer;
    }
}

相关信息

kafka 源码目录

相关文章

kafka AbstractDualSchemaRocksDBSegmentedBytesStore 源码

kafka AbstractMergedSortedCacheStoreIterator 源码

kafka AbstractRocksDBSegmentedBytesStore 源码

kafka AbstractRocksDBTimeOrderedSegmentedBytesStore 源码

kafka AbstractSegments 源码

kafka AbstractStoreBuilder 源码

kafka BatchWritingStore 源码

kafka BlockBasedTableConfigWithAccessibleCache 源码

kafka BufferKey 源码

kafka BufferValue 源码