expect always earliest offset on empty initial offset for the records

[lanwen]

Pulsar by default caches the result of the seek, so in case you do seek,
then create new subscription with the same group name without seek and
earliest subscription type, it will get back to the last seeked offset
what is different from the Kafka and should be aligned in the liiklus

[lanwen]

Tried this:

.delaySubscription(
        initialOffset.map(PulsarRecordsStorage::fromOffset)
                .switchIfEmpty(Mono.justOrEmpty(autoOffsetReset
                        .map(it -> {
                            switch (it) {
                                case "earliest":
                                    return MessageId.earliest;
                                case "latest":
                                    return MessageId.latest;
                                default:
                                    return null;
                            }
                        })
                ))
                .defaultIfEmpty(MessageId.earliest)
                .flatMap(messageId -> {
                    return Mono.fromCompletionStage(consumer.seekAsync(messageId));
                })
);

it fixes the test, but fails with testMultipleGroups and testExclusiveRecordDistribution

[lanwen]

with this fix, testExclusiveRecordDistribution seems fails because after seek it sticks to the first subscription doing that and completely ignores another one, so regardless of the partition all new records will go to the first one

[lanwen]

@bsideup disabled the test for pulsar, so that PR could be merged on its own