[feat][txn] Transaction buffer snapshot writer reuse

[gaoran10]

Motivation

The transaction buffer(short for TB) snapshot topic is used to persistent snapshot data for TB, the snapshot can reduce read data when recovering.

Currently, if enable the transaction feature, every topic will create a TB snapshot producer, the topic of the producer is a namespace system topic, so we don't need to create a producer for every topic, one producer per namespace is enough.

Modifications

Add a new inner class ReferenceCountedWriter, if the ReferenceCountedWriter for one namespace does not exist, it will be initialized, or else its reference count will be increased, if the reference count reduces to 0, it will be removed from map-cache.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: gaoran10#23

[poorbarcode]

Great improve, left some comments

[poorbarcode]

Why not change the param TopicName to Namespace?

[gaoran10]

fixed

[congbobo184]

Should we lock in release and retain? Otherwise, it may result in getting a writer whose referenceCount is 0

[liangyepianzhou]

We may not only need to lock release and retain.
If we do not lock release and retain.

time	task execute ReferenceCountedWriter::close	task execute getReferenceWriter
0	check referenceCount != 0	retain check referenceCount != 0
1	decrement referenceCount = 0
2	remove and close the writer
3		increment referenceCount
4		return the closed writer

[liangyepianzhou]

If we only lock release and retain.

time	task execute ReferenceCountedWriter::close	task execute getReferenceWriter
0	check referenceCount != 0
1	decrement referenceCount = 0
2	remove and close the writer
3		retain check referenceCount == 0
4		throw RuntimeException

[liangyepianzhou]

We can change the retain method to

        private synchronized boolean retain() {
            if (referenceCount.get() == 0) {
                return false;
            } else {
                this.referenceCount.incrementAndGet();
                return true;
            }
        }

And change getReferenceWriter to

    public ReferenceCountedWriter<T> getReferenceWriter(TopicName topicName) {
        return refCountedWriterMap.compute(topicName.getNamespaceObject(), (k, v) -> {
            if (v != null && v.retain()) {
                return v;
            } else {
                return new ReferenceCountedWriter<>(topicName.getNamespaceObject(),
                        getTransactionBufferSystemTopicClient(topicName).newWriterAsync(), this);

            }
        });
    }

[gaoran10]

Good catch! I adjust logic.

[liangyepianzhou]

This method is useless. We can delete it, right?

[gaoran10]

fixed

[liangyepianzhou]

We may not only need to lock release and retain.
If we do not lock release and retain.

time	task execute ReferenceCountedWriter::close	task execute getReferenceWriter
0	check referenceCount != 0	retain check referenceCount != 0
1	decrement referenceCount = 0
2	remove and close the writer
3		increment referenceCount
4		return the closed writer

[liangyepianzhou]

If we only lock release and retain.

time	task execute ReferenceCountedWriter::close	task execute getReferenceWriter
0	check referenceCount != 0
1	decrement referenceCount = 0
2	remove and close the writer
3		retain check referenceCount == 0
4		throw RuntimeException

[liangyepianzhou]

We can change the retain method to

        private synchronized boolean retain() {
            if (referenceCount.get() == 0) {
                return false;
            } else {
                this.referenceCount.incrementAndGet();
                return true;
            }
        }

And change getReferenceWriter to

    public ReferenceCountedWriter<T> getReferenceWriter(TopicName topicName) {
        return refCountedWriterMap.compute(topicName.getNamespaceObject(), (k, v) -> {
            if (v != null && v.retain()) {
                return v;
            } else {
                return new ReferenceCountedWriter<>(topicName.getNamespaceObject(),
                        getTransactionBufferSystemTopicClient(topicName).newWriterAsync(), this);

            }
        });
    }

[gaoran10]

@liangyepianzhou Good idea! I make some changes, PLTA.

[gaoran10]

An error was encountered while deleting the namespace, I'm trying to fix it.

[dao-jun]

Please rebase master branch, the testBatchMetadataMetrics should be fixed