dekaf: Fix high watermark reporting

[jshearer]

Description:

We are returning an incorrectly low high-watermark on the first chunk of a Read. This is unexpected behavior and causes the consumer to be incorrectly behind by 1 fetch chunk (the first one). This should fix the problem going forward, though some affected consumer groups may need to be reset (the equivalent of a re-backfill).

For the collection that this was reported in:

$ flowctl collections read --uncommitted --collection ... | jq -c 'select(.["_meta"]?.ack != true)' | jq -s 'map(.id) | sort_by(.)[]' | uniq > unique_flowctl_ids.txt

$ kcat -o beginning  -b dekaf.estuary-data.com:9092 \                                                                                                                                                                                                                                                                       
-X security.protocol=SASL_SSL \      
-X sasl.mechanism=PLAIN \
-X sasl.username='{}' \             
-X sasl.password="ey.." \
-t ... \
-s value=avro \
-r \ 'https://{}:[email protected]'  -J -C -e | |jq -R "fromjson? | ." | jq -c 'select(.payload.id) | .payload.id' | jq -s -c 'sort_by(.)[]' | uniq > unique_dekaf_ids.txt

$ cat unique_dekaf_ids.txt| | wc -l                                                                                                                                                                                                                                                                                                                      
  680726
$ cat unique_flowctl_ids.txt| uniq | wc -l                                                                                                                                                                                                                                                                                                                    
  686939

... a little while later, testing the fix locally ...

$ kcat -o beginning -b localhost:22262 \                                                                                                                                                                                                                                                                                
-X security.protocol=SASL_PLAINTEXT \
-X sasl.mechanism=PLAIN \
-X sasl.username='{}' \
-X sasl.password="ey.." \
-t ... \
-s value=avro \
-r 'http://{}:ey..@localhost:9081'  -J -C -e | |jq -R "fromjson? | ." | jq -c 'select(.payload.id) | .payload.id' | jq -s -c 'sort_by(.)[]' | uniq > dekaf_fixed_ids.txt

$ cat dekaf_fixed_ids.txt|  wc -l                                                                                                                                                                                                                                                                                                                       
  686954

---

This change is 

[jgraettinger]

LGTM