[zjxxzjwang]

[fix][tableview] Pass nil to listeners on tombstone messages instead of the zero value

Motivation

TableViewImpl.handleMessage passes the deserialized payload to every registered
listener. For tombstone messages (empty payload), payload is created via
reflect.New(SchemaValueType) but is never populated by GetSchemaValue,
so listeners end up receiving the type's zero value
(e.g. MyStruct{}, "", 0, *T(nil-indirect)).

This has two concrete problems:

1. Ambiguous semantics. Listeners cannot distinguish between:

   • a key being deleted (tombstone), and
   • a key being updated to the zero value of its schema type.

   Any downstream logic that reacts to deletions (cache eviction, state
   machine transitions, metrics) is forced to also consult tv.data
   under its own lock, which is both racy and unergonomic.

2. Inconsistent with the Java client. The Java TableViewImpl
   explicitly delivers null to listeners on tombstones. The Go client
   silently diverged from that contract.

Minimal reproducer before the fix:

tv.ForEachAndListen(func(k string, v interface{}) error {
    // v is MyStruct{} both when "k" is set to MyStruct{} AND when "k" is
    // deleted via an empty-payload message. There is no way to tell them apart.
    return nil
})

Modifications

• In pulsar/table_view_impl.go, refactored handleMessage to compute a single
  value variable and pass it to all listeners:
  • Tombstone branch (len(msg.Payload()) == 0): delete the key from
    tv.data and set value = nil.
  • Normal branch: deserialize into a fresh payload, set
    value = reflect.Indirect(payload).Interface(), and store it in tv.data.
• Listener invocation now uses value instead of re-reading an
  unconditionally-created payload, so tombstones consistently deliver nil.
• No change to tv.data contents, lock scope, or any other public method.

Verifying this change

This change added tests and can be verified as follows:

• Added TestHandleMessageListenerValue in pulsar/table_view_test.go.
  It is a pure unit test (no broker required) that drives handleMessage
  directly and asserts the listener contract on three cases:
  1. Normal message → listener receives the deserialized value and
     tv.data is updated accordingly.
  2. Tombstone on an existing key → listener receives nil (not the
     type zero value) and the key is removed from tv.data.
  3. Tombstone on a key that never existed → listener still receives nil.
• The existing integration test TestPublishNilValue continues to pass,
  confirming that end-to-end tombstone handling against a real broker is
  unaffected.

Run locally:

go test -run TestHandleMessageListenerValue -count=1 -v ./pulsar/

• Make sure that the change passes the CI checks.

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

• Dependencies (does it add or upgrade a dependency): no
• The public API: yes (behavioral change, not a signature change) —
  listeners registered via ForEachAndListen now receive nil instead of
  the schema type's zero value when a tombstone (empty-payload message)
  arrives. Callers that previously relied on receiving a zero value
  without a nil check must update their listener to handle nil as a
  deletion signal. This brings the Go client in line with the Java
  TableViewImpl contract.
• The schema: no
• The default values of configurations: no
• The wire protocol: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable
• If a feature is not applicable for documentation, explain why?
  This is a bug fix that aligns the Go TableView listener semantics
  with the already-documented Java client behavior. The GoDoc on
  ForEachAndListen can optionally be updated in a follow-up to call out
  the nil convention for tombstones.
• If a feature is not documented yet in this PR, please create a followup
  issue for adding the documentation: n/a

[zjxxzjwang]

@crossoverJie Hello, can you help me review this PR.