[improve][fn] make built-in connector reload incremental

[Dream95]

Motivation

reloadBuiltInSources / reloadBuiltInSinks currently reload connectors by rebuilding the full connector map and closing all existing connector instances.
This causes unnecessary classloader churn and can invalidate previously referenced connector instances even when the archive content has not changed.

Modifications

• Added incremental reload logic in ConnectorUtils.reloadConnectors(...):
  • Reuse existing Connector instances when both archive path and archive MD5 are unchanged.
  • Create new Connector instances only for added/changed archives.
  • Return evicted connectors (removed/replaced) so callers can close only those.
• Updated ConnectorsManager.reloadConnectors(...) to use the incremental reload result and close only evicted connectors.
• Extended Connector to keep archive MD5 (archiveMd5Hex) for identity checks during reload.
• Extracted MD5 utility methods into pulsar-common FileUtils and reused them from both connector reload and NarUnpacker.
• Added tests:
  • ConnectorUtilsReloadTest
    • unchanged archive is reused and remains usable
    • changed archive is replaced and old connector is closed
    • removed archive is evicted and old connector is closed
  • ConnectorsManagerReloadConnectorsTest
    • manager-level reload reuses unchanged connector instance

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• Added unit tests in pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/io/ConnectorUtilsReloadTest.java
• Added runtime-level test in pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/worker/ConnectorsManagerReloadConnectorsTest.java
• CI should cover compilation and execution of the new tests.

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

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Matching PR in forked repository

PR in forked repository: Dream95#9

[lhotari]

good work. some minor comments

[lhotari]

good work. some minor comments

[dao-jun]

ConnectorsManager.connectors is volatile, and reloadConnectors assigns a new map atomically. The closeConnectors is called on only the evicted set. Between the assignment this.connectors = reload.getLeft() and the close of evicted connectors, there's a brief window where a caller could be using an evicted connector.
However, this was already the case before the change (old behavior also swapped the map then closed the old one), so this PR doesn't regress thread safety. Worth noting but not a new concern.

[Dream95]

/pulsarbot rerun-failure-checks

[lhotari]

LGTM.

The existing code (before this PR) unnecessarily uses TreeMap as parameter, field, or variable types. A general recommendation is to use the minimal Map interface type, usually one of java.util.Map, java.util.SequencedMap, or java.util.SortedMap. In the case of connectors, the use the TreeMap is justified as the implementation to get stable iteration order by key. This is available when java.util.Map is the parameter, field, or variable type.

I'd suggest addressing the refactoring to get rid of unnecessary use of TreeMap in parameter, field, and variable types for connector related classes after this current PR has been merged. For example,

In ConnectorsManager

    @Getter
    private volatile Map<String, Connector> connectors;

The implementation instance would remain as TreeMap, so this change would only be about the types for fields, variables and parameters.

public record ReloadConnectorsResult(Map<String, Connector> connectors, List<Connector> connectorsToClose) {
}

This change could also be made part of this PR since it already closes touches the code where TreeMap is used and also adds a new TreeMap parameter/field in ReloadConnectorsResult.

[lhotari]

@Dream95 There's #25782 about the follow up related to TreeMap types. Would you like to take it?

[Dream95]

@lhotari Yes, I'll take separate PR. Besides the connector changes here, FunctionsManager has the same unnecessary TreeMap usage; I'll address both together.

[Dream95]

Since reload is admin-driven and usually serialized, If we need stricter ordering, I’m happy to add synchronization in a follow-up.

[lhotari]

LGTM