[ISSUE #6560] Avoid junk Kafka log records during producer initialization.

[sunnysabor]

Fixes #6560

What changed

• Replace the synthetic initialization record with a Kafka producer topic metadata lookup.
• Treat metadata lookup failure as initialization failure and close the producer.
• Add regression tests for successful metadata lookup without a send and for metadata timeout cleanup.

Compatibility and impact

The plugin no longer writes a synthetic record to the configured log topic. Initialization now waits for topic metadata and reports authentication, authorization, or metadata timeout errors synchronously. Normal request-log publishing is unchanged.

Tests

• Passed: ./mvnw -pl shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-kafka -am -Dtest=KafkaLogCollectClientTest -Dsurefire.failIfNoSpecifiedTests=false -DskipRemoteResources=true test (2 tests, 0 failures; Checkstyle passed). A temporary Maven Central settings file was used because this machine's default mirror was unavailable.

Checklist

• Read the contribution guidelines.
• Added regression tests for the changed behavior.

[Aias00]

Suggestion (non-blocking): this call blocks until metadata arrives or max.block.ms elapses (60 s by default), and initClient0 is reached from LoggingKafkaPluginDataHandler#doRefreshConfig, i.e. the config-sync path - an unreachable broker would park it for up to a minute per refresh.

The previous send() was not better here (KafkaProducer#send calls waitOnMetadata with the same budget), so this is not a regression, just worth bounding now that fetching metadata is the whole point of the probe. Since kafka-clients 3.9.2 has no partitionsFor(String, Duration) overload (verified with javap against the exact version this reactor depends on), the only lever is:

props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 5_000L);

Also worth considering: when the topic does not exist and auto-creation is disabled, partitionsFor simply returns an empty list instead of raising, so a mistyped topic still reports a successful init.

[Aias00]

Approving - this is a real fix, not cosmetic.

Why it matters (checked against the call chain): GrpcPlugin#doExecute passes extInfo.methodType to ShenyuGrpcClient#call, whose invoke(...) does switch (methodType), and JsonMessage#getMethodType switches on it as well. Both throw NullPointerException for a null methodType. Since Gson's fromJson(String, Class) returns null for both null and "", any grpc metadata whose rpcExt was absent or blank crashed with an NPE inside the plugin chain - exactly #6650. Handling null/blank and giving GrpcExtInfo a non-null default therefore removes a genuine crash path, and the three new tests (null / "" / " ") plus asserting the parsed method type are the coverage that was missing. Using a fixed {"timeout":5000,"methodType":"SERVER_STREAMING"} fixture instead of the previous lenient {timeout:5000} is also better hygiene.

Two follow-ups (non-blocking, happy either way):

1. The guard covers whole-object null only. If rpcExt parses but leaves methodType null - an unknown enum name resolves to null in Gson, and "methodType": null sets it explicitly - the field is still null and invoke(...) NPEs again with a null methodType. Worth normalising after parsing (if getMethodType() == null fall back to UNARY), which would close the same class of bug completely.

2. Defaulting to UNARY is the sensible choice (it beats crashing), but it silently assumes unary semantics for metadata that forgot to declare a method type: a mis-registered server-streaming method now gets a confusing downstream result instead of the immediate NPE. A LOG.warn when the defaults are used would keep that diagnosable. Related nit: a {"timeout":0} payload produces a 0 ms deadline - consider keeping the 5000 default when the parsed value is null or <= 0.

CI is green across build, integrated tests and e2e, so nothing here blocks merge.