private Stream createStream() throws ExecutionException, InterruptedException {
        CreateStreamOptions.Builder options = CreateStreamOptions.builder().replicaCount(replicaCount)
            .epoch(epoch);
        tags.forEach(options::tag);
        return client.createAndOpenStream(options.build()).get();
    }
Lazy stream, create stream when append record.
core/src/main/scala/kafka/log/streamaspect/LazyStream.java