|
| 1 | +/* |
| 2 | + * Licensed to the Apache Software Foundation (ASF) under one |
| 3 | + * or more contributor license agreements. See the NOTICE file |
| 4 | + * distributed with this work for additional information |
| 5 | + * regarding copyright ownership. The ASF licenses this file |
| 6 | + * to you under the Apache License, Version 2.0 (the |
| 7 | + * "License"); you may not use this file except in compliance |
| 8 | + * with the License. You may obtain a copy of the License at |
| 9 | + * |
| 10 | + * http://www.apache.org/licenses/LICENSE-2.0 |
| 11 | + * |
| 12 | + * Unless required by applicable law or agreed to in writing, |
| 13 | + * software distributed under the License is distributed on an |
| 14 | + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| 15 | + * KIND, either express or implied. See the License for the |
| 16 | + * specific language governing permissions and limitations |
| 17 | + * under the License. |
| 18 | + */ |
| 19 | +package org.apache.pulsar.broker.service; |
| 20 | + |
| 21 | +import java.util.ArrayList; |
| 22 | +import java.util.List; |
| 23 | +import java.util.Set; |
| 24 | +import java.util.UUID; |
| 25 | +import lombok.CustomLog; |
| 26 | +import org.apache.pulsar.broker.PulsarService; |
| 27 | +import org.apache.pulsar.client.admin.PulsarAdmin; |
| 28 | +import org.apache.pulsar.client.api.PulsarClient; |
| 29 | +import org.apache.pulsar.client.api.PulsarClientException; |
| 30 | +import org.testng.annotations.AfterMethod; |
| 31 | +import org.testng.annotations.BeforeClass; |
| 32 | +import org.testng.annotations.BeforeMethod; |
| 33 | + |
| 34 | +/** |
| 35 | + * Base class for tests that need a shared multi-broker cluster across test classes. |
| 36 | + * |
| 37 | + * <p>Companion to {@link SharedPulsarBaseTest}. Use this when a test specifically depends on |
| 38 | + * behavior that only manifests across brokers — namespace ownership transfer, controller-leader |
| 39 | + * failover, segment placement on different brokers, V5 client reconnect to a different broker. |
| 40 | + * For everything else, prefer the single-broker {@link SharedPulsarBaseTest}: it's faster, has |
| 41 | + * fewer moving parts, and is sufficient for most coverage. |
| 42 | + * |
| 43 | + * <p>Each test method gets a fresh namespace under {@link SharedMultiBrokerPulsarCluster#TENANT_NAME} |
| 44 | + * (created in {@link #setupSharedMultiBrokerTest()} and force-deleted in |
| 45 | + * {@link #cleanupSharedMultiBrokerTest()}). The cluster itself is JVM-wide and reused across |
| 46 | + * every test class that extends this — see {@link SharedMultiBrokerPulsarCluster}. |
| 47 | + * |
| 48 | + * <p>Subclasses get: |
| 49 | + * <ul> |
| 50 | + * <li>{@link #admin} / {@link #pulsarClient} — handles aimed at broker 0; lookups against any |
| 51 | + * broker correctly redirect to topic owners, so most tests should just use these.</li> |
| 52 | + * <li>{@link #brokers} / {@link #admins} / {@link #clients} — full per-broker lists, in start |
| 53 | + * order, for tests that need to address a specific broker (e.g. asserting topic |
| 54 | + * ownership, killing a specific broker).</li> |
| 55 | + * <li>{@link #newTopicName()} — generates a unique topic in the test namespace.</li> |
| 56 | + * </ul> |
| 57 | + */ |
| 58 | +@CustomLog |
| 59 | +public abstract class SharedMultiBrokerPulsarBaseTest { |
| 60 | + |
| 61 | + /** All brokers in the shared cluster, in start order. */ |
| 62 | + protected List<PulsarService> brokers; |
| 63 | + /** Per-broker admin handles, aligned with {@link #brokers}. */ |
| 64 | + protected List<PulsarAdmin> admins; |
| 65 | + /** Per-broker client handles, aligned with {@link #brokers}. */ |
| 66 | + protected List<PulsarClient> clients; |
| 67 | + |
| 68 | + /** Convenience: broker 0's admin. */ |
| 69 | + protected PulsarAdmin admin; |
| 70 | + /** Convenience: broker 0's client. */ |
| 71 | + protected PulsarClient pulsarClient; |
| 72 | + |
| 73 | + private final List<String> namespaces = new ArrayList<>(); |
| 74 | + |
| 75 | + /** Returns the unique namespace assigned to the current test method. */ |
| 76 | + protected String getNamespace() { |
| 77 | + return namespaces.get(0); |
| 78 | + } |
| 79 | + |
| 80 | + /** Returns the broker service URL for broker {@code index}. */ |
| 81 | + protected String getBrokerServiceUrl(int index) { |
| 82 | + return brokers.get(index).getBrokerServiceUrl(); |
| 83 | + } |
| 84 | + |
| 85 | + /** Returns the web service URL for broker {@code index}. */ |
| 86 | + protected String getWebServiceUrl(int index) { |
| 87 | + return brokers.get(index).getWebServiceAddress(); |
| 88 | + } |
| 89 | + |
| 90 | + /** |
| 91 | + * Creates a new {@link PulsarClient} connected to broker {@code index}. Callers are |
| 92 | + * responsible for closing the returned client. |
| 93 | + */ |
| 94 | + protected PulsarClient newPulsarClient(int index) throws PulsarClientException { |
| 95 | + return PulsarClient.builder().serviceUrl(brokers.get(index).getBrokerServiceUrl()).build(); |
| 96 | + } |
| 97 | + |
| 98 | + /** |
| 99 | + * Initializes (lazily) the shared cluster singleton and wires the per-class fields. Called |
| 100 | + * once per test class. |
| 101 | + */ |
| 102 | + @BeforeClass(alwaysRun = true) |
| 103 | + public void setupSharedMultiBrokerCluster() throws Exception { |
| 104 | + SharedMultiBrokerPulsarCluster cluster = SharedMultiBrokerPulsarCluster.get(); |
| 105 | + brokers = cluster.getBrokers(); |
| 106 | + admins = cluster.getAdmins(); |
| 107 | + clients = cluster.getClients(); |
| 108 | + admin = cluster.getAdmin(); |
| 109 | + pulsarClient = cluster.getClient(); |
| 110 | + } |
| 111 | + |
| 112 | + /** Creates a unique namespace for the current test method. */ |
| 113 | + @BeforeMethod(alwaysRun = true) |
| 114 | + public void setupSharedMultiBrokerTest() throws Exception { |
| 115 | + createNewNamespace(); |
| 116 | + } |
| 117 | + |
| 118 | + /** Force-deletes all namespaces created during the test method. */ |
| 119 | + @AfterMethod(alwaysRun = true) |
| 120 | + public void cleanupSharedMultiBrokerTest() throws Exception { |
| 121 | + for (String ns : namespaces) { |
| 122 | + try { |
| 123 | + admin.namespaces().deleteNamespace(ns, true); |
| 124 | + log.info().attr("testNamespace", ns).log("Deleted test namespace"); |
| 125 | + } catch (Exception e) { |
| 126 | + log.warn().attr("deleteNamespace", ns).exceptionMessage(e).log("Failed to delete namespace"); |
| 127 | + } |
| 128 | + } |
| 129 | + namespaces.clear(); |
| 130 | + } |
| 131 | + |
| 132 | + /** Creates a new namespace under the shared tenant and registers it for cleanup. */ |
| 133 | + protected String createNewNamespace() throws Exception { |
| 134 | + String nsName = "test-" + UUID.randomUUID().toString().substring(0, 8); |
| 135 | + String ns = SharedMultiBrokerPulsarCluster.TENANT_NAME + "/" + nsName; |
| 136 | + admin.namespaces().createNamespace(ns, Set.of(SharedMultiBrokerPulsarCluster.CLUSTER_NAME)); |
| 137 | + namespaces.add(ns); |
| 138 | + log.info().attr("testNamespace", ns).log("Created test namespace"); |
| 139 | + return ns; |
| 140 | + } |
| 141 | + |
| 142 | + /** Generates a unique persistent topic name within the current test namespace. */ |
| 143 | + protected String newTopicName() { |
| 144 | + return "persistent://" + getNamespace() + "/topic-" + UUID.randomUUID().toString().substring(0, 8); |
| 145 | + } |
| 146 | +} |
0 commit comments