public static void applyChecksAndUpdatePosition(
            final ConsumerRecord<byte[], byte[]> record,
            final boolean consistencyEnabled,
            final Position position
    ) {
        if (!consistencyEnabled) {
            return;
        }
        final Header versionHeader = record.headers().lastHeader(
                ChangelogRecordDeserializationHelper.CHANGELOG_VERSION_HEADER_KEY);
        if (versionHeader == null) {
            return;
        } else {
            switch (versionHeader.value()[0]) {
                case 0:
                    final Header vectorHeader = record.headers().lastHeader(CHANGELOG_POSITION_HEADER_KEY);
                    if (vectorHeader == null) {
                        throw new StreamsException("This should not happen. Consistency is enabled but the changelog "
                                + "contains records without consistency information.");
                    }
                    position.merge(PositionSerde.deserialize(ByteBuffer.wrap(vectorHeader.value())));
                    break;
                default:
                    log.warn("Changelog records have been encoded using a larger version than this server understands." +
                            "Please upgrade your server.");
            }
        }
    }
Changelog records without any headers are considered old format.
New format changelog records will have a version in their headers.
Version 0: This indicates that the changelog records have consistency information.
streams/src/main/java/org/apache/kafka/streams/processor/internals/ChangelogRecordDeserializationHelper.java