logical streaming of xacts via test_decoding is broken

Michael reported a BF failure [1] related to one of the logical
streaming test case and I've analyzed the issue. As responded on
pgsql-committers [2], the issue here is that the streaming
transactions can be interleaved and because we are maintaining whether
xact_wrote_changes at the LogicalDecodingContext level, one of later
transaction can overwrite the flag for previously streaming
transaction. I think it is logical to have this flag at each
transaction level (aka in ReorderBufferTxn), however till now it was
fine because the changes of each transaction are decoded at one-shot
which will be no longer true. We can keep a output_plugin_private data
pointer in ReorderBufferTxn which will be used by test_decoding module
to keep this and any other such flags in future. We need to set this
flag at begin_cb and stream_start_cb APIs and then reset/remove it at
stream_commit_cb, stream_abort_cb and stream_stop_cb APIs.

Additionally, we can extend the existing test case
concurrent_stream.spec to cover this scenario by adding a step to have
an empty transaction before the commit of transaction which we are
going to stream changes for (before s1_commit).

Thoughts?

[1] - https://www.postgresql.org/message-id/20201109014118.GD1695%40paquier.xyz
[2] - https://www.postgresql.org/message-id/CAA4eK1JMCm9HURVmOapo%2Bv2u2EEABOuzgp7XJ32C072ygcKktQ%40mail.gmail.com

--
With Regards,
Amit Kapila.