diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java index e5e69ea14ad32..f10b2cbdc322e 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java @@ -3834,6 +3834,22 @@ public double getLoadBalancerBandwidthOutResourceWeight() { ) private int transactionCoordinatorScalableTopicsGcRetentionSeconds = 900; + @FieldContext( + category = CATEGORY_TRANSACTION, + minValue = 1, + doc = "Degree of parallelism for the scalable-topics transaction coordinator: how many" + + " independent coordinator instances run across the cluster. Each is" + + " leader-elected independently in the metadata store and coordinates the" + + " transactions whose id maps to it. Fixed at cluster bring-up — changing it" + + " later would strand the coordinator id encoded in existing transaction ids" + + " (and, because an aborted transaction's records are retained as long as its" + + " messages are, the value can only be reduced once all transactions created" + + " under the previous value have been fully cleaned up). All brokers must agree" + + " on this value; a mismatch is rejected at startup. Only relevant when" + + " transactionCoordinatorScalableTopicsEnabled = true." + ) + private int transactionCoordinatorScalableTopicsParallelism = 16; + @FieldContext( category = CATEGORY_TRANSACTION, doc = "Class name for transaction metadata store provider" diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java index af7e8930f36dd..00253bbb4311d 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java @@ -108,6 +108,7 @@ import org.apache.pulsar.broker.service.schema.exceptions.InvalidSchemaDataException; import org.apache.pulsar.broker.topiclistlimit.TopicListMemoryLimiter; import org.apache.pulsar.broker.topiclistlimit.TopicListSizeResultCache; +import org.apache.pulsar.broker.transaction.coordinator.v5.TransactionCoordinatorV5; import org.apache.pulsar.broker.web.RestException; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.transaction.TxnID; @@ -505,6 +506,10 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception { }); scalableTopicsWatchers.clear(); + // Same for transaction-coordinator assignment watchers. + tcAssignmentWatchers.values().forEach(this::closeQuietly); + tcAssignmentWatchers.clear(); + // Notify the scalable-topic controller that this connection's scalable consumers // have dropped. The controller marks them disconnected and starts the grace-period // timer; if they reconnect in time, their assignment is preserved. @@ -865,6 +870,13 @@ protected void handleCommandScalableTopicLookup( ScalableTopicsWatcherSession> scalableTopicsWatchers = new ConcurrentHashMap<>(); + // --- Transaction-coordinator assignment watchers --- + // watchId -> deregistration handle for the listener registered on TransactionCoordinatorV5. + private final ConcurrentHashMap tcAssignmentWatchers = new ConcurrentHashMap<>(); + // Delay before re-pushing a TC-assignment snapshot that was incomplete (a partition mid-election) + // or that failed to build, so the client converges without waiting for an external trigger. + private static final long TC_ASSIGNMENTS_REPUSH_DELAY_MS = 1000L; + @Override protected void handleCommandWatchScalableTopics( CommandWatchScalableTopics cmd) { @@ -965,6 +977,85 @@ protected void handleCommandWatchScalableTopicsClose( } } + // --- Transaction-coordinator assignment watch --- + + @Override + protected void handleCommandWatchTcAssignments( + org.apache.pulsar.common.api.proto.CommandWatchTcAssignments cmd) { + checkArgument(state == State.Connected); + final long watchId = cmd.getWatchId(); + log.debug().attr("watchId", watchId).log("Received WatchTcAssignments"); + + if (!service.getPulsar().getConfig().isTransactionCoordinatorScalableTopicsEnabled()) { + ctx.writeAndFlush(Commands.newWatchTcAssignmentsError(watchId, ServerError.NotAllowedError, + "Scalable-topics transaction coordinator is disabled on this broker")); + return; + } + TransactionCoordinatorV5 tc = service.getPulsar().getTransactionCoordinatorV5(); + if (tc == null) { + ctx.writeAndFlush(Commands.newWatchTcAssignmentsError(watchId, ServerError.ServiceNotReady, + "Transaction coordinator not ready")); + return; + } + // Register a listener that re-pushes the full snapshot on any leadership change, then send + // the initial snapshot. Authz: this is broker-internal coordination, not a per-topic op, so + // an authenticated connection is sufficient (same trust model as TC_CLIENT_CONNECT). + AutoCloseable handle = tc.registerAssignmentChangeListener( + () -> ctx.executor().execute(() -> sendTcAssignmentsSnapshot(watchId, tc))); + AutoCloseable prev = tcAssignmentWatchers.put(watchId, handle); + closeQuietly(prev); + sendTcAssignmentsSnapshot(watchId, tc); + } + + private void sendTcAssignmentsSnapshot(long watchId, TransactionCoordinatorV5 tc) { + if (!tcAssignmentWatchers.containsKey(watchId)) { + return; + } + tc.buildAssignmentsSnapshot().thenAccept(snapshot -> ctx.executor().execute(() -> { + if (!tcAssignmentWatchers.containsKey(watchId)) { + return; + } + java.util.Map leaders = new java.util.HashMap<>(); + snapshot.assignments().forEach((partition, leader) -> leaders.put(partition, + new String[] {leader.brokerServiceUrl(), leader.brokerServiceUrlTls()})); + ctx.writeAndFlush(Commands.newWatchTcAssignmentsSnapshot( + watchId, snapshot.partitionCount(), leaders)); + // If some partition is still mid-election, the snapshot is incomplete. Schedule a single + // delayed re-push so the client doesn't stay parked on a missing partition waiting for a + // leadership change that may never come (the cache repopulating fires no TC listener). + if (!snapshot.isComplete()) { + ctx.executor().schedule(() -> sendTcAssignmentsSnapshot(watchId, tc), + TC_ASSIGNMENTS_REPUSH_DELAY_MS, TimeUnit.MILLISECONDS); + } + })).exceptionally(ex -> { + log.warn().attr("watchId", watchId).exception(ex) + .log("Failed to build TC-assignments snapshot; retrying shortly"); + ctx.executor().schedule(() -> sendTcAssignmentsSnapshot(watchId, tc), + TC_ASSIGNMENTS_REPUSH_DELAY_MS, TimeUnit.MILLISECONDS); + return null; + }); + } + + @Override + protected void handleCommandWatchTcAssignmentsClose( + org.apache.pulsar.common.api.proto.CommandWatchTcAssignmentsClose cmd) { + checkArgument(state == State.Connected); + long watchId = cmd.getWatchId(); + log.debug().attr("watchId", watchId).log("Received WatchTcAssignmentsClose"); + closeQuietly(tcAssignmentWatchers.remove(watchId)); + } + + private void closeQuietly(AutoCloseable handle) { + if (handle == null) { + return; + } + try { + handle.close(); + } catch (Exception e) { + log.warn().exceptionMessage(e).log("Error closing TC-assignment watcher"); + } + } + @Override protected void handleCommandScalableTopicClose( CommandScalableTopicClose commandScalableTopicClose) { @@ -1316,7 +1407,8 @@ private void completeConnect(int clientProtoVersion, String clientVersion) { maybeScheduleAuthenticationCredentialsRefresh(); } writeAndFlush(Commands.newConnected(clientProtoVersion, maxMessageSize, enableTopicListWatcher, - scalableTopicsEnabled)); + scalableTopicsEnabled, + service.getPulsar().getConfig().isTransactionCoordinatorScalableTopicsEnabled())); state = State.Connected; service.getPulsarStats().recordConnectionCreateSuccess(); log.debug() diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/coordinator/v5/TransactionCoordinatorV5.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/coordinator/v5/TransactionCoordinatorV5.java index d4bfd8581a270..22ad4aee3ef85 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/coordinator/v5/TransactionCoordinatorV5.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/coordinator/v5/TransactionCoordinatorV5.java @@ -24,9 +24,13 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.Set; +import java.util.TreeMap; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentSkipListMap; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @@ -35,6 +39,7 @@ import lombok.CustomLog; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.transaction.exception.coordinator.TransactionCoordinatorException; +import org.apache.pulsar.broker.transaction.metadata.TcLeader; import org.apache.pulsar.broker.transaction.metadata.TxnEvent; import org.apache.pulsar.broker.transaction.metadata.TxnHeader; import org.apache.pulsar.broker.transaction.metadata.TxnIds; @@ -50,6 +55,8 @@ import org.apache.pulsar.common.util.FutureUtil; import org.apache.pulsar.metadata.api.GetResult; import org.apache.pulsar.metadata.api.ScanConsumer; +import org.apache.pulsar.metadata.api.coordination.LeaderElection; +import org.apache.pulsar.metadata.api.coordination.LeaderElectionState; import org.apache.pulsar.transaction.coordinator.TransactionCoordinatorID; import org.apache.pulsar.transaction.coordinator.TransactionSubscription; import org.apache.pulsar.transaction.coordinator.exceptions.CoordinatorException; @@ -57,10 +64,20 @@ /** * Metadata-driven transaction coordinator for scalable topics — broker-side service. * - *

Per-partition coordinator. A broker runs the TC for partition {@code N} iff it owns - * partition {@code N} of {@code SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN} — same - * leader-election mechanism the legacy {@code TransactionMetadataStoreService} uses; reusing - * it keeps the client-side discovery surface unchanged. + *

Per-partition coordinator. Leadership rests on the metadata store directly: each TC + * partition {@code N} has a {@link LeaderElection} at {@code /txn/tc/leader/}, and a broker + * runs the TC for partition {@code N} iff it currently leads that election. This removes the + * dependency on the {@code transaction_coordinator_assign} topic and its bundle ownership — TC + * coordination liveness no longer rides on the topic/namespace/load-balancer machinery, only on + * the metadata store (which the TC already hard-depends on for every header read/write). + * + *

Distribution. Every broker calls {@code elect()} on every partition (elect-all): + * the {@code LeaderElection} primitive only fails a leader over to a broker that is already a + * candidate, so to keep every partition survivable every broker must be a candidate for every + * partition. The N independent elections start concurrently, so on a co-start leadership lands + * roughly balanced across brokers, and every partition has B−1 standby candidates for instant + * failover. (After a strictly sequential scale-up an early broker can hold more partitions until + * it restarts; TC load is light, so v1 does not actively rebalance.) * *

Wire commands handled (routed by {@code ServerCnx} when * {@code transactionCoordinatorScalableTopicsEnabled} is on): @@ -71,6 +88,8 @@ * advertise themselves by writing {@code /txn/op} records, so the TC doesn't need a * pre-registration step. *

  • {@code END_TXN} → {@link #endTransaction}
    • + *
  • {@code WATCH_TC_ASSIGNMENTS} → {@link #buildAssignmentsSnapshot} + push-on-change, the + * client's discovery surface (which broker leads which partition).
    • * * *

    {@code endTransaction} CAS-updates the header to the terminal state, enumerates @@ -79,18 +98,18 @@ * {@code (segment, subscription)} pair. The fan-out is metadata-store writes (not RPCs) and * is bounded by the txn's participant count. * - *

    Background sweeps: a single elected broker — the owner of partition 0 of - * {@code transaction_coordinator_assign} — periodically (a) aborts timed-out open transactions - * ({@link #sweepTimeouts}) and (b) garbage-collects finalized transactions whose retention has - * elapsed ({@link #sweepGc}). Concurrent sweeps from a stale owner are still safe — every state - * transition is a header CAS — so the single-sweeper election is an efficiency measure, not a - * correctness one. + *

    Background sweeps: the broker that leads partition 0 periodically (a) aborts timed-out open + * transactions ({@link #sweepTimeouts}) and (b) garbage-collects finalized transactions whose + * retention has elapsed ({@link #sweepGc}). Concurrent sweeps from a stale leader are still safe + * — every state transition is a header CAS — so the single-sweeper election is an efficiency + * measure, not a correctness one. */ @CustomLog public class TransactionCoordinatorV5 { private final PulsarService pulsar; private final TxnMetadataStore txnStore; + private final int partitionCount; private final long timeoutSweepIntervalMs; private final long gcSweepIntervalMs; @@ -100,10 +119,18 @@ public class TransactionCoordinatorV5 { private final AtomicBoolean timeoutSweepRunning = new AtomicBoolean(false); private final AtomicBoolean gcSweepRunning = new AtomicBoolean(false); + /** Per-partition leader-election controllers, keyed by partition (0..partitionCount-1). */ + private final Map> elections = new ConcurrentHashMap<>(); + /** The local broker's election value — what we propose for every partition we lead. */ + private volatile TcLeader localLeader; + /** Open assignment-watch listeners (one per watching client connection). */ + private final List assignmentChangeListeners = new CopyOnWriteArrayList<>(); + public TransactionCoordinatorV5(PulsarService pulsar) { this.pulsar = pulsar; this.txnStore = new TxnMetadataStore(pulsar.getLocalMetadataStore()); var config = pulsar.getConfiguration(); + this.partitionCount = config.getTransactionCoordinatorScalableTopicsParallelism(); this.timeoutSweepIntervalMs = TimeUnit.SECONDS.toMillis( config.getTransactionCoordinatorScalableTopicsTimeoutSweepIntervalSeconds()); this.gcSweepIntervalMs = TimeUnit.SECONDS.toMillis( @@ -115,14 +142,32 @@ public TransactionCoordinatorV5(PulsarService pulsar) { // ---- Lifecycle -------------------------------------------------------- /** - * Start the periodic timeout / GC sweeps on a dedicated single-thread scheduler. Each tick is - * gated by {@link #ifElectedSweeper} so only the partition-0 owner does the scan. Idempotent — - * a second call is ignored. + * Start the coordinator: create a per-partition {@link LeaderElection} and {@code elect()} on + * every partition (elect-all), then start the periodic timeout / GC sweeps on a dedicated + * single-thread scheduler. Sweep ticks are gated by {@link #ifElectedSweeper} so only the + * partition-0 leader scans. Idempotent — a second call is ignored. */ public synchronized void start() { if (closed || sweepExecutor != null) { return; } + verifyParallelismConsistency(); + this.localLeader = new TcLeader(pulsar.getBrokerId(), pulsar.getBrokerServiceUrl(), + pulsar.getBrokerServiceUrlTls(), pulsar.getSafeWebServiceAddress()); + for (int partition = 0; partition < partitionCount; partition++) { + final int p = partition; + LeaderElection election = pulsar.getCoordinationService().getLeaderElection( + TcLeader.class, TxnPaths.tcLeaderPath(p), state -> onElectionStateChange(p, state)); + elections.put(p, election); + // elect-all: become a candidate for every partition so leadership is balanced across + // brokers and every partition has standbys for failover. Errors are logged; the + // LeaderElection retries internally. + election.elect(localLeader).exceptionally(ex -> { + log.warn().attr("partition", p).exception(ex).log("v5 TC initial elect failed"); + return null; + }); + } + sweepExecutor = Executors.newSingleThreadScheduledExecutor( new DefaultThreadFactory("pulsar-txn-v5-sweep")); sweepExecutor.scheduleWithFixedDelay( @@ -133,13 +178,87 @@ public synchronized void start() { gcSweepIntervalMs, gcSweepIntervalMs, TimeUnit.MILLISECONDS); } - /** Stop the sweeps. Idempotent. */ + /** + * Persist this broker's configured parallelism cluster-wide on first start, and verify every + * subsequent broker agrees. A mismatch means brokers would run different election sets and the + * coordinator-count encoded in transaction ids would be ambiguous — fatal misconfiguration, so + * we fail fast rather than start in an inconsistent state. + */ + private void verifyParallelismConsistency() { + var store = pulsar.getLocalMetadataStore(); + try { + byte[] value = Integer.toString(partitionCount).getBytes(java.nio.charset.StandardCharsets.UTF_8); + var existing = store.get(TxnPaths.TXN_TC_PARALLELISM_PATH).get(); + if (existing.isEmpty()) { + // First broker to start writes the value (CAS create; lose harmlessly to a racing peer). + store.put(TxnPaths.TXN_TC_PARALLELISM_PATH, value, java.util.Optional.of(-1L)) + .get(); + var after = store.get(TxnPaths.TXN_TC_PARALLELISM_PATH).get(); + if (after.isPresent()) { + checkParallelismMatches(after.get().getValue()); + } + } else { + checkParallelismMatches(existing.get().getValue()); + } + } catch (IllegalStateException e) { + throw e; + } catch (Exception e) { + // A racing create (BadVersion) or read-after-write resolves by re-reading and comparing. + try { + var after = store.get(TxnPaths.TXN_TC_PARALLELISM_PATH).get(); + after.ifPresent(r -> checkParallelismMatches(r.getValue())); + } catch (Exception ignore) { + log.warn().exception(e).log("Could not verify TC parallelism consistency; proceeding"); + } + } + } + + private void checkParallelismMatches(byte[] storedValue) { + int stored = Integer.parseInt(new String(storedValue, java.nio.charset.StandardCharsets.UTF_8).trim()); + if (stored != partitionCount) { + throw new IllegalStateException( + "transactionCoordinatorScalableTopicsParallelism mismatch: this broker is configured" + + " with " + partitionCount + " but the cluster was initialized with " + stored + + ". The value is fixed at cluster bring-up and must be identical on every" + + " broker."); + } + } + + /** Stop the sweeps and release every leader-election lease. Idempotent. */ public synchronized void close() { closed = true; if (sweepExecutor != null) { sweepExecutor.shutdownNow(); sweepExecutor = null; } + elections.values().forEach(e -> e.asyncClose().exceptionally(ex -> { + log.warn().exception(ex).log("v5 TC election close failed"); + return null; + })); + elections.clear(); + assignmentChangeListeners.clear(); + } + + /** + * Whether this broker currently leads TC partition {@code partition}. Used to gate + * client-connect acceptance and the sweep. A partition with no local election (out-of-range or + * pre-{@code start()}) is not led here. + */ + public boolean isLeaderFor(int partition) { + LeaderElection election = elections.get(partition); + return election != null && election.getState() == LeaderElectionState.Leading; + } + + /** Fire every assignment-watch listener — a leader changed somewhere, so the map moved. */ + private void onElectionStateChange(int partition, LeaderElectionState state) { + log.debug().attr("partition", partition).attr("state", state).log("v5 TC election state changed"); + for (Runnable listener : assignmentChangeListeners) { + try { + listener.run(); + } catch (Throwable t) { + log.warn().exception(t).log("v5 TC assignment listener failed"); + } + } } /** @@ -174,17 +293,75 @@ private void runSweep(String name, AtomicBoolean running, Supplier handleClientConnect(TransactionCoordinatorID tcId) { + if (isLeaderFor((int) tcId.getId())) { + return CompletableFuture.completedFuture(null); + } String assignPartition = SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN .getPartition((int) tcId.getId()).toString(); return pulsar.getBrokerService().checkTopicNsOwnership(assignPartition); } + // ---- Assignment discovery (client watch) ------------------------------ + + /** + * Build the current full {@code partition → leader} snapshot from the election state. Uses the + * async {@link LeaderElection#getLeaderValue()} (which loads from the metadata store on a cache + * miss) rather than the cache-only {@code getLeaderValueIfPresent()}: when this broker just + * transitioned to {@code Following} for a partition, its local cache for the new leader's node + * may not be repopulated yet, and a cache-only read would silently omit that partition. Loading + * from the store closes that window so a follower's snapshot is still complete. + * + *

    A partition still genuinely without a leader (no broker elected yet) is omitted; the caller + * ({@code ServerCnx}) re-pushes shortly after so the client isn't stranded. Always the complete + * map — the watch protocol sends full snapshots, never diffs. + * + * @return a future of the snapshot plus whether it is complete (every partition has a leader) + */ + public CompletableFuture buildAssignmentsSnapshot() { + Map assignments = new ConcurrentSkipListMap<>(); + List> loads = new ArrayList<>(elections.size()); + for (Map.Entry> e : elections.entrySet()) { + int partition = e.getKey(); + loads.add(e.getValue().getLeaderValue() + .thenAccept(opt -> opt.ifPresent(leader -> assignments.put(partition, leader))) + .exceptionally(ex -> { + // Treat a load error as "leader unknown for now"; the re-push will retry. + log.debug().attr("partition", partition).exception(ex) + .log("v5 TC leader-value load failed while building snapshot"); + return null; + })); + } + return FutureUtil.waitForAll(loads) + .thenApply(__ -> new TcAssignmentsSnapshot(partitionCount, new TreeMap<>(assignments))); + } + + /** + * Register a listener fired whenever the assignment map may have changed (any partition's + * leadership moved). Returns an {@link AutoCloseable} that deregisters it — the + * {@code ServerCnx} closes it when the client closes the watch or disconnects. + */ + public AutoCloseable registerAssignmentChangeListener(Runnable listener) { + assignmentChangeListeners.add(listener); + return () -> assignmentChangeListeners.remove(listener); + } + + /** Immutable full assignment snapshot: partition count + the currently-known leaders. */ + public record TcAssignmentsSnapshot(int partitionCount, Map assignments) { + /** @return true if every partition has a known leader (no mid-election gaps). */ + public boolean isComplete() { + return assignments.size() == partitionCount; + } + } + // ---- newTransaction --------------------------------------------------- /** @@ -459,21 +636,15 @@ public void onCompleted() { } /** - * Run {@code action} only on the elected sweeper — the broker that owns partition 0 of - * {@code transaction_coordinator_assign}. Not owning it (or any error checking ownership) means - * "skip this cycle". Correctness doesn't depend on the election: every transition is a header - * CAS, so a stale owner sweeping concurrently is harmless. + * Run {@code action} only on the elected sweeper — the broker that leads TC partition 0. Not + * leading it means "skip this cycle". Correctness doesn't depend on the election: every + * transition is a header CAS, so a stale leader sweeping concurrently is harmless. */ private CompletableFuture ifElectedSweeper(Supplier> action) { - if (closed) { + if (closed || !isLeaderFor(0)) { return CompletableFuture.completedFuture(null); } - String assignPartition0 = SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN - .getPartition(0).toString(); - return pulsar.getBrokerService().checkTopicNsOwnership(assignPartition0) - .handle((v, ex) -> ex == null) - .thenCompose(owned -> (owned && !closed) - ? action.get() : CompletableFuture.completedFuture(null)); + return action.get(); } /** A {@code (segment, subscription)} ack participant; keys the ack fan-out de-dup set. */ diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/metadata/TcLeader.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/metadata/TcLeader.java new file mode 100644 index 0000000000000..cefba83d7c99c --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/metadata/TcLeader.java @@ -0,0 +1,39 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.transaction.metadata; + +/** + * Value stored in a per-partition transaction-coordinator leader-election node + * ({@code /txn/tc/leader/}). Identifies the broker currently coordinating that TC + * partition and carries the connection URLs a client needs to reach it — so any broker can + * answer a client's assignment watch from the election value alone, without a further lookup. + * + *

    Serialized as JSON via {@link org.apache.pulsar.common.util.ObjectMapperFactory} by the + * coordination service's {@code LeaderElection} serde. + * + * @param brokerId the elected broker's id (matches the {@code /loadbalance/brokers} key) + * @param brokerServiceUrl the broker's binary service URL (non-TLS); may be {@code null} if the + * broker only advertises a TLS endpoint + * @param brokerServiceUrlTls the broker's binary service URL (TLS); may be {@code null} if TLS is + * disabled + * @param webServiceUrl the broker's HTTP service URL, for admin/CLI resolution + */ +public record TcLeader(String brokerId, String brokerServiceUrl, String brokerServiceUrlTls, + String webServiceUrl) { +} diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/metadata/TxnPaths.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/metadata/TxnPaths.java index 0c5bbcfd97219..e94d0b1705395 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/metadata/TxnPaths.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/metadata/TxnPaths.java @@ -107,6 +107,28 @@ public static String tcSequencePath(long tcId) { return TXN_TC_SEQ_PREFIX + "/" + tcId; } + /** + * Path prefix for the per-partition transaction-coordinator leader-election nodes. Each + * partition {@code N} has a {@code LeaderElection} under {@code /txn/tc/leader/} whose + * value is the {@link TcLeader} currently coordinating that partition. Replaces the + * {@code transaction_coordinator_assign} topic as the v5 TC's election surface — election + * rests on the metadata store directly, not on topic/bundle ownership. + */ + public static final String TXN_TC_LEADER_PREFIX = "/txn/tc/leader"; + + /** @return {@code /txn/tc/leader/} — the leader-election node for {@code partition}. */ + public static String tcLeaderPath(int partition) { + return TXN_TC_LEADER_PREFIX + "/" + partition; + } + + /** + * Cluster-wide record of the scalable-topics TC parallelism, written once by the first broker to + * start. Every broker verifies its configured value against this and refuses to start on a + * mismatch, so the coordinator-count encoded in transaction ids stays stable for the cluster's + * lifetime. + */ + public static final String TXN_TC_PARALLELISM_PATH = "/txn/tc/parallelism"; + /** Width used when formatting long values into lexicographically-orderable index keys. */ public static final int LONG_KEY_WIDTH = 20; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/coordinator/v5/TransactionCoordinatorV5Test.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/coordinator/v5/TransactionCoordinatorV5Test.java index c5f261e85aec0..9db63021dd983 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/coordinator/v5/TransactionCoordinatorV5Test.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/coordinator/v5/TransactionCoordinatorV5Test.java @@ -39,7 +39,9 @@ import org.apache.pulsar.client.api.transaction.TxnID; import org.apache.pulsar.common.api.proto.TxnAction; import org.apache.pulsar.metadata.api.MetadataStoreConfig; +import org.apache.pulsar.metadata.api.coordination.CoordinationService; import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended; +import org.apache.pulsar.metadata.coordination.impl.CoordinationServiceImpl; import org.apache.pulsar.transaction.coordinator.TransactionCoordinatorID; import org.apache.pulsar.transaction.coordinator.exceptions.CoordinatorException; import org.awaitility.Awaitility; @@ -60,6 +62,7 @@ public class TransactionCoordinatorV5Test { private TxnMetadataStore txnStore; private PulsarService pulsar; private BrokerService brokerService; + private CoordinationService coordinationService; private TransactionCoordinatorV5 tc; @BeforeMethod @@ -67,17 +70,30 @@ public void setUp() throws Exception { store = MetadataStoreExtended.create("memory:local", MetadataStoreConfig.builder().fsyncEnable(false).build()); txnStore = new TxnMetadataStore(store); + coordinationService = new CoordinationServiceImpl(store); pulsar = mock(PulsarService.class); when(pulsar.getLocalMetadataStore()).thenReturn(store); + when(pulsar.getCoordinationService()).thenReturn(coordinationService); + when(pulsar.getBrokerId()).thenReturn("broker-test:8080"); + when(pulsar.getBrokerServiceUrl()).thenReturn("pulsar://broker-test:6650"); + when(pulsar.getBrokerServiceUrlTls()).thenReturn(null); + when(pulsar.getSafeWebServiceAddress()).thenReturn("http://broker-test:8080"); ServiceConfiguration cfg = new ServiceConfiguration(); // GC sweep tests assume retention has already elapsed. cfg.setTransactionCoordinatorScalableTopicsGcRetentionSeconds(0); + // Keep the election small so start() converges quickly in unit tests. + cfg.setTransactionCoordinatorScalableTopicsParallelism(4); when(pulsar.getConfiguration()).thenReturn(cfg); brokerService = mock(BrokerService.class); when(pulsar.getBrokerService()).thenReturn(brokerService); - // Default: owned. Tests that want to assert the not-owned path can override. + // Default: owned (assign-topic fallback path in handleClientConnect). Tests that want to + // assert the not-owned path can override. when(brokerService.checkTopicNsOwnership(any())).thenReturn(CompletableFuture.completedFuture(null)); tc = new TransactionCoordinatorV5(pulsar); + tc.start(); + // As the only broker, we win every partition's election; wait until partition 0 is led so + // the sweep-gating and client-connect paths behave deterministically. + Awaitility.await().until(() -> tc.isLeaderFor(0)); } @AfterMethod(alwaysRun = true) @@ -85,6 +101,9 @@ public void tearDown() throws Exception { if (tc != null) { tc.close(); } + if (coordinationService != null) { + coordinationService.close(); + } if (store != null) { store.close(); } @@ -320,16 +339,89 @@ public void sweepGc_repairsAndRetainsHeaderWhenOpsRemain() throws Exception { } @Test - public void sweeps_skipWhenNotElected() throws Exception { - // Override the owned-default with a failure → not the elected sweeper → action skipped. - when(brokerService.checkTopicNsOwnership(any())).thenReturn( - CompletableFuture.failedFuture(new RuntimeException("not owner"))); - + public void sweeps_skipWhenNotLeader() throws Exception { + // Create an expired txn, then drop leadership (close releases the election leases). The + // sweep is gated on isLeaderFor(0), so on a fresh non-leader TC it must skip. TxnID expired = tc.newTransaction(TC_ID, 1L, "owner").get(); - tc.sweepTimeouts().get(); + tc.close(); + + // A second TC that never started (no elections) is not the leader for partition 0. + TransactionCoordinatorV5 notLeader = new TransactionCoordinatorV5(pulsar); + try { + notLeader.sweepTimeouts().get(); + // Still OPEN — the sweep never ran because this TC leads no partition. + var header = txnStore.getHeader(TxnIds.toKey(expired)).get().orElseThrow(); + assertThat(header.value().getState()).isEqualTo(TxnState.OPEN); + } finally { + notLeader.close(); + } + } - // Still OPEN — the sweep never ran because we don't own assign-partition 0. - var header = txnStore.getHeader(TxnIds.toKey(expired)).get().orElseThrow(); - assertThat(header.value().getState()).isEqualTo(TxnState.OPEN); + // ---- Election + assignment discovery ---------------------------------- + + @Test + public void election_singleBrokerLeadsAllPartitions() { + // As the only broker, this TC wins every partition's election. + for (int p = 0; p < 4; p++) { + final int partition = p; + Awaitility.await().until(() -> tc.isLeaderFor(partition)); + } + assertThat(tc.isLeaderFor(4)).isFalse(); // out of range (parallelism = 4) + } + + @Test + public void buildAssignmentsSnapshot_reportsAllLedPartitions() { + assertThat(tc.buildAssignmentsSnapshot().join().partitionCount()).isEqualTo(4); + Awaitility.await().untilAsserted(() -> { + var snap = tc.buildAssignmentsSnapshot().join(); + assertThat(snap.assignments()).hasSize(4); + assertThat(snap.isComplete()).isTrue(); + assertThat(snap.assignments().get(0).brokerServiceUrl()) + .isEqualTo("pulsar://broker-test:6650"); + assertThat(snap.assignments().get(0).brokerId()).isEqualTo("broker-test:8080"); + }); + } + + @Test + public void registerAssignmentChangeListener_deregistersOnClose() throws Exception { + // The handle deregisters the listener; after close() it must not be invoked again. We only + // assert the registration/deregistration contract here — the fire-on-election-change path + // needs a multi-broker setup and is covered at integration level. + AutoCloseable handle = tc.registerAssignmentChangeListener(() -> { }); + handle.close(); + } + + @Test + public void handleClientConnect_acceptsWhenLeader() throws Exception { + // We lead partition 0, so connect is accepted without consulting assign-topic ownership. + tc.handleClientConnect(TC_ID).get(); + } + + @Test + public void start_failsOnParallelismMismatch() throws Exception { + // The running tc (from setUp) persisted parallelism=4. A second coordinator configured with a + // different value against the same metadata store must refuse to start. + ServiceConfiguration mismatchCfg = new ServiceConfiguration(); + mismatchCfg.setTransactionCoordinatorScalableTopicsGcRetentionSeconds(0); + mismatchCfg.setTransactionCoordinatorScalableTopicsParallelism(8); + PulsarService other = mock(PulsarService.class); + when(other.getLocalMetadataStore()).thenReturn(store); + when(other.getCoordinationService()).thenReturn(coordinationService); + when(other.getConfiguration()).thenReturn(mismatchCfg); + when(other.getBrokerId()).thenReturn("broker-other:8080"); + when(other.getBrokerServiceUrl()).thenReturn("pulsar://broker-other:6650"); + when(other.getSafeWebServiceAddress()).thenReturn("http://broker-other:8080"); + when(other.getBrokerService()).thenReturn(brokerService); + + TransactionCoordinatorV5 mismatched = new TransactionCoordinatorV5(other); + try { + assertThatThrownBy(mismatched::start) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("mismatch") + .hasMessageContaining("8") + .hasMessageContaining("4"); + } finally { + mismatched.close(); + } } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionClientConnectTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionClientConnectTest.java index ecd0479d085cc..e63542cd06040 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionClientConnectTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionClientConnectTest.java @@ -138,10 +138,7 @@ public void testTransactionAddPublishPartitionToTxnReconnect() throws Exception @Test public void testPulsarClientCloseThenCloseTcClient() throws Exception { TransactionCoordinatorClientImpl transactionCoordinatorClient = ((PulsarClientImpl) pulsarClient).getTcClient(); - Field field = TransactionCoordinatorClientImpl.class.getDeclaredField("handlers"); - field.setAccessible(true); - TransactionMetaStoreHandler[] handlers = - (TransactionMetaStoreHandler[]) field.get(transactionCoordinatorClient); + java.util.Collection handlers = transactionCoordinatorClient.getHandlers(); for (TransactionMetaStoreHandler handler : handlers) { handler.newTransactionAsync(10, TimeUnit.SECONDS).get(); @@ -168,11 +165,8 @@ public void testPulsarClientCloseThenCloseTcClient() throws Exception { public void testHandlerStateChangeToReady() throws Exception { TransactionCoordinatorClientImpl transactionCoordinatorClient = ((PulsarClientImpl) pulsarClient).getTcClient(); - Field field = TransactionCoordinatorClientImpl.class.getDeclaredField("handlers"); - field.setAccessible(true); - TransactionMetaStoreHandler[] handlers = - (TransactionMetaStoreHandler[]) field.get(transactionCoordinatorClient); - TransactionMetaStoreHandler transactionMetaStoreHandler = handlers[0]; + TransactionMetaStoreHandler transactionMetaStoreHandler = + transactionCoordinatorClient.getHandlers().iterator().next(); Assert.assertEquals(transactionMetaStoreHandler.getConnectHandleState(), HandlerState.State.Ready); Assert.assertTrue(transactionMetaStoreHandler.changeToReadyState()); } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java index ff12f01b40fba..16d78585531a6 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java @@ -190,6 +190,12 @@ public class ClientCnx extends PulsarHandler { .concurrencyLevel(1) .build(); + private final ConcurrentLongHashMap tcAssignmentsWatchers = + ConcurrentLongHashMap.newBuilder() + .expectedItems(2) + .concurrencyLevel(1) + .build(); + private final CompletableFuture connectionFuture = new CompletableFuture(); private final ConcurrentLinkedQueue requestTimeoutQueue = new ConcurrentLinkedQueue<>(); @@ -238,6 +244,8 @@ public class ClientCnx extends PulsarHandler { private boolean supportsTopicWatcherReconcile; @Getter private boolean supportsScalableTopics; + @Getter + private boolean supportsTcMetadataDiscovery; /** Idle stat. **/ @Getter @@ -393,6 +401,7 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception { dagWatchSessions.forEach((__, session) -> session.connectionClosed()); scalableConsumerSessions.forEach((__, session) -> session.connectionClosed()); scalableTopicsWatchers.forEach((__, session) -> session.connectionClosed()); + tcAssignmentsWatchers.forEach((__, session) -> session.connectionClosed()); waitingLookupRequests.clear(); @@ -402,6 +411,7 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception { dagWatchSessions.clear(); scalableConsumerSessions.clear(); scalableTopicsWatchers.clear(); + tcAssignmentsWatchers.clear(); timeoutTask.cancel(true); } @@ -458,6 +468,8 @@ protected void handleConnected(CommandConnected connected) { connected.hasFeatureFlags() && connected.getFeatureFlags().isSupportsTopicWatcherReconcile(); supportsScalableTopics = connected.hasFeatureFlags() && connected.getFeatureFlags().isSupportsScalableTopics(); + supportsTcMetadataDiscovery = + connected.hasFeatureFlags() && connected.getFeatureFlags().isSupportsTcMetadataDiscovery(); // set remote protocol version to the correct version before we complete the connection future setRemoteEndpointProtocolVersion(connected.getProtocolVersion()); @@ -1516,6 +1528,62 @@ public void removeScalableTopicsWatcher(long watchId) { scalableTopicsWatchers.remove(watchId); } + /** Client-side receiver for transaction-coordinator assignment snapshots. */ + public interface TcAssignmentsWatcherSession { + void onSnapshot(int parallelism, java.util.Map leaders); + + void onError(org.apache.pulsar.common.api.proto.ServerError error, String message); + + void connectionClosed(); + } + + public void registerTcAssignmentsWatcher(long watchId, TcAssignmentsWatcherSession watcher) { + tcAssignmentsWatchers.put(watchId, watcher); + } + + public void removeTcAssignmentsWatcher(long watchId) { + tcAssignmentsWatchers.remove(watchId); + } + + @Override + protected void handleCommandWatchTcAssignmentsUpdate( + org.apache.pulsar.common.api.proto.CommandWatchTcAssignmentsUpdate cmd) { + checkArgument(state == State.Ready); + long watchId = cmd.getWatchId(); + log.debug().attr("watchId", watchId).log("Received WatchTcAssignmentsUpdate"); + + if (cmd.hasError()) { + TcAssignmentsWatcherSession session = tcAssignmentsWatchers.remove(watchId); + if (session != null) { + session.onError(cmd.getError(), cmd.hasMessage() ? cmd.getMessage() : null); + } else { + log.warn().attr("watchId", watchId) + .log("Received TC-assignments watch error for unknown watcher"); + } + return; + } + + TcAssignmentsWatcherSession session = tcAssignmentsWatchers.get(watchId); + if (session == null) { + log.warn().attr("watchId", watchId) + .log("Received TC-assignments watch update for unknown watcher"); + return; + } + if (!cmd.hasSnapshot()) { + log.warn().attr("watchId", watchId).log("TC-assignments update with no snapshot payload"); + return; + } + var snapshot = cmd.getSnapshot(); + java.util.Map leaders = new java.util.HashMap<>(); + for (int i = 0; i < snapshot.getAssignmentsCount(); i++) { + var a = snapshot.getAssignmentAt(i); + leaders.put(a.getTcId(), new String[] { + a.hasBrokerServiceUrl() ? a.getBrokerServiceUrl() : null, + a.hasBrokerServiceUrlTls() ? a.getBrokerServiceUrlTls() : null}); + } + session.onSnapshot(snapshot.getParallelism(), leaders); + } + /** * check serverError and take appropriate action. *