[fix][transaction] Use pulsar format when write marker to __consumer_offsets topic

[Demogorgon314]

Motivation

We should use pulsar format to encode the marker when writing into the __consumer_offsets topic, otherwise it will have the following errors:

2023-08-09T09:17:18,398+0000 [pulsar-client-io-69-3] ERROR org.apache.pulsar.client.impl.ConsumerImpl - [public/__kafka/__consumer_offsets-partition-42][reader-ca0ccccd79] Discarding corrupted message at 6:5
2023-08-09T09:17:18,402+0000 [pulsar-client-io-69-3] WARN  org.apache.pulsar.client.impl.ConsumerImpl - [reader-ca0ccccd79] [98942] unable to obtain message in batch
java.lang.IllegalStateException: java.lang.IllegalStateException: Some required fields are missing

Modifications

Use pulsar format when write marker to __consumer_offsets topic

Documentation

Check the box below.

Need to update docs?

• [ ] doc-required

  (If you need help on updating docs, create a doc issue)

• [x] no-need-doc

  (Please explain why)

• [ ] doc

  (If this PR contains doc changes)

[codecov[bot]]

Codecov Report

Merging #1994 (a09be4b) into master (26980a3) will increase coverage by 0.00%. Report is 1 commits behind head on master. The diff coverage is 71.42%.

@@            Coverage Diff            @@
##             master    #1994   +/-   ##
=========================================
  Coverage     17.05%   17.06%           
  Complexity      728      728           
=========================================
  Files           191      191           
  Lines         14235    14239    +4     
  Branches       1332     1334    +2     
=========================================
+ Hits           2428     2430    +2     
- Misses        11631    11633    +2     
  Partials        176      176           

Files Changed	Coverage Δ
...tive/pulsar/handlers/kop/storage/PartitionLog.java	8.34% <0.00%> (-0.02%)	:arrow_down:
...sar/handlers/kop/format/EntryFormatterFactory.java	61.53% <100.00%> (ø)
...ative/pulsar/handlers/kop/utils/MetadataUtils.java	71.97% <100.00%> (+0.36%)	:arrow_up:

... and 1 file with indirect coverage changes