| /* |
| * 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.tests.integration.topologies; |
| |
| import static com.google.common.base.Preconditions.checkArgument; |
| import static org.apache.pulsar.tests.integration.containers.PulsarContainer.BROKER_HTTPS_PORT; |
| import static org.apache.pulsar.tests.integration.containers.PulsarContainer.BROKER_HTTP_PORT; |
| import static org.apache.pulsar.tests.integration.containers.PulsarContainer.BROKER_PORT_TLS; |
| import static org.apache.pulsar.tests.integration.containers.PulsarContainer.CS_PORT; |
| import static org.apache.pulsar.tests.integration.containers.PulsarContainer.PULSAR_CONTAINERS_LEAVE_RUNNING; |
| import static org.apache.pulsar.tests.integration.containers.PulsarContainer.ZK_PORT; |
| import com.google.common.collect.Lists; |
| import com.google.common.collect.Maps; |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Objects; |
| import java.util.function.Function; |
| import lombok.Getter; |
| import lombok.extern.slf4j.Slf4j; |
| import org.apache.commons.io.IOUtils; |
| import org.apache.pulsar.client.impl.auth.AuthenticationTls; |
| import org.apache.pulsar.tests.integration.containers.BKContainer; |
| import org.apache.pulsar.tests.integration.containers.BrokerContainer; |
| import org.apache.pulsar.tests.integration.containers.CSContainer; |
| import org.apache.pulsar.tests.integration.containers.ProxyContainer; |
| import org.apache.pulsar.tests.integration.containers.PulsarContainer; |
| import org.apache.pulsar.tests.integration.containers.WorkerContainer; |
| import org.apache.pulsar.tests.integration.containers.ZKContainer; |
| import org.apache.pulsar.tests.integration.docker.ContainerExecResult; |
| import org.testcontainers.containers.BindMode; |
| import org.testcontainers.containers.GenericContainer; |
| import org.testcontainers.containers.Network; |
| |
| /** |
| * Pulsar Cluster in containers. |
| */ |
| @Slf4j |
| public class PulsarCluster { |
| |
| public static final String ADMIN_SCRIPT = "/pulsar/bin/pulsar-admin"; |
| public static final String CLIENT_SCRIPT = "/pulsar/bin/pulsar-client"; |
| public static final String PULSAR_COMMAND_SCRIPT = "/pulsar/bin/pulsar"; |
| public static final String CURL = "/usr/bin/curl"; |
| |
| /** |
| * Pulsar Cluster Spec |
| * |
| * @param spec pulsar cluster spec. |
| * @return the built pulsar cluster |
| */ |
| public static PulsarCluster forSpec(PulsarClusterSpec spec) { |
| CSContainer csContainer = new CSContainer(spec.clusterName) |
| .withNetwork(Network.newNetwork()) |
| .withNetworkAliases(CSContainer.NAME); |
| return new PulsarCluster(spec, csContainer, false); |
| } |
| |
| public static PulsarCluster forSpec(PulsarClusterSpec spec, CSContainer csContainer) { |
| return new PulsarCluster(spec, csContainer, true); |
| } |
| |
| @Getter |
| private final PulsarClusterSpec spec; |
| |
| @Getter |
| private final String clusterName; |
| private final Network network; |
| private final ZKContainer zkContainer; |
| private final CSContainer csContainer; |
| private final boolean sharedCsContainer; |
| private final Map<String, BKContainer> bookieContainers; |
| private final Map<String, BrokerContainer> brokerContainers; |
| private final Map<String, WorkerContainer> workerContainers; |
| private final ProxyContainer proxyContainer; |
| private Map<String, GenericContainer<?>> externalServices = Collections.emptyMap(); |
| private Map<String, Map<String, String>> externalServiceEnvs; |
| |
| private PulsarCluster(PulsarClusterSpec spec, CSContainer csContainer, boolean sharedCsContainer) { |
| |
| this.spec = spec; |
| this.sharedCsContainer = sharedCsContainer; |
| this.clusterName = spec.clusterName(); |
| this.network = csContainer.getNetwork(); |
| |
| this.zkContainer = new ZKContainer(clusterName); |
| this.zkContainer |
| .withNetwork(network) |
| .withNetworkAliases(appendClusterName(ZKContainer.NAME)) |
| .withEnv("clusterName", clusterName) |
| .withEnv("zkServers", appendClusterName(ZKContainer.NAME)) |
| .withEnv("configurationStore", CSContainer.NAME + ":" + CS_PORT) |
| .withEnv("forceSync", "no") |
| .withEnv("pulsarNode", appendClusterName("pulsar-broker-0")); |
| |
| this.csContainer = csContainer; |
| |
| this.bookieContainers = Maps.newTreeMap(); |
| this.brokerContainers = Maps.newTreeMap(); |
| this.workerContainers = Maps.newTreeMap(); |
| |
| this.proxyContainer = new ProxyContainer(appendClusterName("pulsar-proxy"), ProxyContainer.NAME, spec.enableTls) |
| .withNetwork(network) |
| .withNetworkAliases(appendClusterName("pulsar-proxy")) |
| .withEnv("zkServers", appendClusterName(ZKContainer.NAME)) |
| .withEnv("zookeeperServers", appendClusterName(ZKContainer.NAME)) |
| .withEnv("configurationStoreServers", CSContainer.NAME + ":" + CS_PORT) |
| .withEnv("clusterName", clusterName); |
| // enable mTLS |
| if (spec.enableTls) { |
| proxyContainer |
| .withEnv("webServicePortTls", String.valueOf(BROKER_HTTPS_PORT)) |
| .withEnv("servicePortTls", String.valueOf(BROKER_PORT_TLS)) |
| .withEnv("forwardAuthorizationCredentials", "true") |
| .withEnv("tlsRequireTrustedClientCertOnConnect", "true") |
| .withEnv("tlsAllowInsecureConnection", "false") |
| .withEnv("tlsCertificateFilePath", "/pulsar/certificate-authority/server-keys/proxy.cert.pem") |
| .withEnv("tlsKeyFilePath", "/pulsar/certificate-authority/server-keys/proxy.key-pk8.pem") |
| .withEnv("tlsTrustCertsFilePath", "/pulsar/certificate-authority/certs/ca.cert.pem") |
| .withEnv("brokerClientAuthenticationPlugin", AuthenticationTls.class.getName()) |
| .withEnv("brokerClientAuthenticationParameters", String.format("tlsCertFile:%s,tlsKeyFile:%s", "/pulsar/certificate-authority/client-keys/admin.cert.pem", "/pulsar/certificate-authority/client-keys/admin.key-pk8.pem")) |
| .withEnv("tlsEnabledWithBroker", "true") |
| .withEnv("brokerClientTrustCertsFilePath", "/pulsar/certificate-authority/certs/ca.cert.pem") |
| .withEnv("brokerClientCertificateFilePath", "/pulsar/certificate-authority/server-keys/proxy.cert.pem") |
| .withEnv("brokerClientKeyFilePath", "/pulsar/certificate-authority/server-keys/proxy.key-pk8.pem"); |
| |
| } |
| if (spec.proxyEnvs != null) { |
| spec.proxyEnvs.forEach(this.proxyContainer::withEnv); |
| } |
| if (spec.proxyMountFiles != null) { |
| spec.proxyMountFiles.forEach(this.proxyContainer::withFileSystemBind); |
| } |
| |
| // create bookies |
| bookieContainers.putAll( |
| runNumContainers("bookie", spec.numBookies(), (name) -> { |
| BKContainer bookieContainer = new BKContainer(clusterName, name) |
| .withNetwork(network) |
| .withNetworkAliases(appendClusterName(name)) |
| .withEnv("zkServers", appendClusterName(ZKContainer.NAME)) |
| .withEnv("useHostNameAsBookieID", "true") |
| // Disable fsyncs for tests since they're slow within the containers |
| .withEnv("journalSyncData", "false") |
| .withEnv("journalMaxGroupWaitMSec", "0") |
| .withEnv("clusterName", clusterName) |
| .withEnv("PULSAR_PREFIX_diskUsageWarnThreshold", "0.95") |
| .withEnv("diskUsageThreshold", "0.99") |
| .withEnv("PULSAR_PREFIX_diskUsageLwmThreshold", "0.97") |
| .withEnv("nettyMaxFrameSizeBytes", String.valueOf(spec.maxMessageSize)); |
| if (spec.bookkeeperEnvs != null) { |
| bookieContainer.withEnv(spec.bookkeeperEnvs); |
| } |
| if (spec.bookieAdditionalPorts != null) { |
| spec.bookieAdditionalPorts.forEach(bookieContainer::addExposedPort); |
| } |
| return bookieContainer; |
| }) |
| ); |
| |
| // create brokers |
| brokerContainers.putAll( |
| runNumContainers("broker", spec.numBrokers(), (name) -> { |
| BrokerContainer brokerContainer = new BrokerContainer(clusterName, appendClusterName(name), spec.enableTls) |
| .withNetwork(network) |
| .withNetworkAliases(appendClusterName(name)) |
| .withEnv("zkServers", appendClusterName(ZKContainer.NAME)) |
| .withEnv("zookeeperServers", appendClusterName(ZKContainer.NAME)) |
| .withEnv("configurationStoreServers", CSContainer.NAME + ":" + CS_PORT) |
| .withEnv("clusterName", clusterName) |
| .withEnv("brokerServiceCompactionMonitorIntervalInSeconds", "1") |
| .withEnv("loadBalancerOverrideBrokerNicSpeedGbps", "1") |
| // used in s3 tests |
| .withEnv("AWS_ACCESS_KEY_ID", "accesskey").withEnv("AWS_SECRET_KEY", "secretkey") |
| .withEnv("maxMessageSize", "" + spec.maxMessageSize); |
| if (spec.enableTls) { |
| // enable mTLS |
| brokerContainer |
| .withEnv("webServicePortTls", String.valueOf(BROKER_HTTPS_PORT)) |
| .withEnv("brokerServicePortTls", String.valueOf(BROKER_PORT_TLS)) |
| .withEnv("authenticateOriginalAuthData", "true") |
| .withEnv("tlsAllowInsecureConnection", "false") |
| .withEnv("tlsRequireTrustedClientCertOnConnect", "true") |
| .withEnv("tlsTrustCertsFilePath", "/pulsar/certificate-authority/certs/ca.cert.pem") |
| .withEnv("tlsCertificateFilePath", "/pulsar/certificate-authority/server-keys/broker.cert.pem") |
| .withEnv("tlsKeyFilePath", "/pulsar/certificate-authority/server-keys/broker.key-pk8.pem"); |
| } |
| if (spec.queryLastMessage) { |
| brokerContainer.withEnv("bookkeeperExplicitLacIntervalInMills", "10"); |
| brokerContainer.withEnv("bookkeeperUseV2WireProtocol", "false"); |
| } |
| if (spec.brokerEnvs != null) { |
| brokerContainer.withEnv(spec.brokerEnvs); |
| } |
| if (spec.brokerMountFiles != null) { |
| spec.brokerMountFiles.forEach(brokerContainer::withFileSystemBind); |
| } |
| if (spec.brokerAdditionalPorts() != null) { |
| spec.brokerAdditionalPorts().forEach(brokerContainer::addExposedPort); |
| } |
| return brokerContainer; |
| } |
| )); |
| |
| spec.classPathVolumeMounts.forEach((key, value) -> { |
| zkContainer.withClasspathResourceMapping(key, value, BindMode.READ_WRITE); |
| proxyContainer.withClasspathResourceMapping(key, value, BindMode.READ_WRITE); |
| |
| bookieContainers.values().forEach(c -> c.withClasspathResourceMapping(key, value, BindMode.READ_WRITE)); |
| brokerContainers.values().forEach(c -> c.withClasspathResourceMapping(key, value, BindMode.READ_WRITE)); |
| workerContainers.values().forEach(c -> c.withClasspathResourceMapping(key, value, BindMode.READ_WRITE)); |
| }); |
| |
| } |
| |
| public String getPlainTextServiceUrl() { |
| return proxyContainer.getPlainTextServiceUrl(); |
| } |
| |
| public String getHttpServiceUrl() { |
| return proxyContainer.getHttpServiceUrl(); |
| } |
| |
| public String getAnyBrokersHttpsServiceUrl() { |
| return getAnyBroker().getHttpsServiceUrl(); |
| } |
| |
| public String getAnyBrokersServiceUrlTls() { |
| return getAnyBroker().getServiceUrlTls(); |
| } |
| |
| public String getAllBrokersHttpServiceUrl() { |
| String multiUrl = "http://"; |
| Iterator<BrokerContainer> brokers = getBrokers().iterator(); |
| while (brokers.hasNext()) { |
| BrokerContainer broker = brokers.next(); |
| multiUrl += broker.getHost() + ":" + broker.getMappedPort(BROKER_HTTP_PORT); |
| if (brokers.hasNext()) { |
| multiUrl += ","; |
| } |
| } |
| return multiUrl; |
| } |
| |
| public String getZKConnString() { |
| return zkContainer.getHost() + ":" + zkContainer.getMappedPort(ZK_PORT); |
| } |
| |
| public String getCSConnString() { |
| return csContainer.getHost() + ":" + csContainer.getMappedPort(CS_PORT); |
| } |
| |
| public Network getNetwork() { |
| return network; |
| } |
| |
| public Map<String, GenericContainer<?>> getExternalServices() { |
| return externalServices; |
| } |
| |
| public void start() throws Exception { |
| // start the local zookeeper |
| zkContainer.start(); |
| log.info("Successfully started local zookeeper container."); |
| |
| // start the configuration store |
| if (!sharedCsContainer) { |
| csContainer.start(); |
| log.info("Successfully started configuration store container."); |
| } |
| |
| // init the cluster |
| zkContainer.execCmd( |
| "bin/init-cluster.sh"); |
| log.info("Successfully initialized the cluster."); |
| |
| // start bookies |
| bookieContainers.values().forEach(BKContainer::start); |
| log.info("Successfully started {} bookie containers.", bookieContainers.size()); |
| |
| // start brokers |
| this.startAllBrokers(); |
| log.info("Successfully started {} broker containers.", brokerContainers.size()); |
| |
| // create proxy |
| proxyContainer.start(); |
| log.info("Successfully started pulsar proxy."); |
| |
| log.info("Pulsar cluster {} is up running:", clusterName); |
| log.info("\tBinary Service Url : {}", getPlainTextServiceUrl()); |
| log.info("\tHttp Service Url : {}", getHttpServiceUrl()); |
| |
| // start external services |
| this.externalServices = spec.externalServices; |
| this.externalServiceEnvs = spec.externalServiceEnvs; |
| if (null != externalServices) { |
| externalServices.entrySet().parallelStream().forEach(service -> { |
| GenericContainer<?> serviceContainer = service.getValue(); |
| serviceContainer.withNetwork(network); |
| serviceContainer.withNetworkAliases(service.getKey()); |
| if (null != externalServiceEnvs && null != externalServiceEnvs.get(service.getKey())) { |
| Map<String, String> env = externalServiceEnvs.getOrDefault(service.getKey(), Collections.emptyMap()); |
| serviceContainer.withEnv(env); |
| } |
| PulsarContainer.configureLeaveContainerRunning(serviceContainer); |
| serviceContainer.start(); |
| log.info("Successfully started external service {}.", service.getKey()); |
| }); |
| } |
| } |
| |
| public void startService(String networkAlias, |
| GenericContainer<?> serviceContainer) { |
| log.info("Starting external service {} ...", networkAlias); |
| serviceContainer.withNetwork(network); |
| serviceContainer.withNetworkAliases(networkAlias); |
| PulsarContainer.configureLeaveContainerRunning(serviceContainer); |
| serviceContainer.start(); |
| log.info("Successfully start external service {}", networkAlias); |
| } |
| |
| public static void stopService(String networkAlias, |
| GenericContainer<?> serviceContainer) { |
| if (PULSAR_CONTAINERS_LEAVE_RUNNING) { |
| logIgnoringStopDueToLeaveRunning(); |
| return; |
| } |
| log.info("Stopping external service {} ...", networkAlias); |
| serviceContainer.stop(); |
| log.info("Successfully stop external service {}", networkAlias); |
| } |
| |
| |
| private static <T extends PulsarContainer> Map<String, T> runNumContainers(String serviceName, |
| int numContainers, |
| Function<String, T> containerCreator) { |
| Map<String, T> containers = Maps.newTreeMap(); |
| for (int i = 0; i < numContainers; i++) { |
| String name = "pulsar-" + serviceName + "-" + i; |
| T container = containerCreator.apply(name); |
| containers.put(name, container); |
| } |
| return containers; |
| } |
| |
| public synchronized void stop() { |
| if (PULSAR_CONTAINERS_LEAVE_RUNNING) { |
| logIgnoringStopDueToLeaveRunning(); |
| return; |
| } |
| |
| stopInParallel(workerContainers.values()); |
| |
| if (externalServices != null) { |
| stopInParallel(externalServices.values()); |
| } |
| |
| if (null != proxyContainer) { |
| proxyContainer.stop(); |
| } |
| |
| stopInParallel(brokerContainers.values()); |
| |
| stopInParallel(bookieContainers.values()); |
| |
| if (!sharedCsContainer && null != csContainer) { |
| csContainer.stop(); |
| } |
| |
| if (null != zkContainer) { |
| zkContainer.stop(); |
| } |
| |
| try { |
| network.close(); |
| } catch (Exception e) { |
| log.info("Failed to shutdown network for pulsar cluster {}", clusterName, e); |
| } |
| } |
| |
| private static void stopInParallel(Collection<? extends GenericContainer<?>> containers) { |
| containers.parallelStream() |
| .filter(Objects::nonNull) |
| .forEach(GenericContainer::stop); |
| } |
| |
| public synchronized void setupFunctionWorkers(String suffix, FunctionRuntimeType runtimeType, int numFunctionWorkers) { |
| switch (runtimeType) { |
| case THREAD: |
| startFunctionWorkersWithThreadContainerFactory(suffix, numFunctionWorkers); |
| break; |
| case PROCESS: |
| startFunctionWorkersWithProcessContainerFactory(suffix, numFunctionWorkers); |
| break; |
| } |
| } |
| |
| private void startFunctionWorkersWithProcessContainerFactory(String suffix, int numFunctionWorkers) { |
| workerContainers.putAll(runNumContainers( |
| "functions-worker-process-" + suffix, |
| numFunctionWorkers, |
| (name) -> createWorkerContainer(name) |
| )); |
| this.startWorkers(); |
| } |
| |
| private WorkerContainer createWorkerContainer(String name) { |
| String serviceUrl = "pulsar://pulsar-broker-0:" + PulsarContainer.BROKER_PORT; |
| String httpServiceUrl = "http://pulsar-broker-0:" + PulsarContainer.BROKER_HTTP_PORT; |
| return new WorkerContainer(clusterName, name) |
| .withNetwork(network) |
| .withNetworkAliases(name) |
| // worker settings |
| .withEnv("PF_workerId", name) |
| .withEnv("PF_workerHostname", name) |
| .withEnv("PF_workerPort", "" + PulsarContainer.BROKER_HTTP_PORT) |
| .withEnv("PF_pulsarFunctionsCluster", clusterName) |
| .withEnv("PF_pulsarServiceUrl", serviceUrl) |
| .withEnv("PF_pulsarWebServiceUrl", httpServiceUrl) |
| // script |
| .withEnv("clusterName", clusterName) |
| .withEnv("zookeeperServers", ZKContainer.NAME) |
| // bookkeeper tools |
| .withEnv("zkServers", ZKContainer.NAME); |
| } |
| |
| private void startFunctionWorkersWithThreadContainerFactory(String suffix, int numFunctionWorkers) { |
| workerContainers.putAll(runNumContainers( |
| "functions-worker-thread-" + suffix, |
| numFunctionWorkers, |
| (name) -> createWorkerContainer(name) |
| .withEnv("PF_functionRuntimeFactoryClassName", |
| "org.apache.pulsar.functions.runtime.thread.ThreadRuntimeFactory") |
| .withEnv("PF_functionRuntimeFactoryConfigs_threadGroupName", "pf-container-group") |
| )); |
| this.startWorkers(); |
| } |
| |
| public synchronized void startWorkers() { |
| // Start workers that have been initialized |
| workerContainers.values().parallelStream().forEach(WorkerContainer::start); |
| log.info("Successfully started {} worker containers.", workerContainers.size()); |
| } |
| |
| public synchronized void stopWorker(String workerName) { |
| if (PULSAR_CONTAINERS_LEAVE_RUNNING) { |
| logIgnoringStopDueToLeaveRunning(); |
| return; |
| } |
| // Stop the named worker. |
| WorkerContainer worker = workerContainers.get(workerName); |
| if (worker == null) { |
| log.warn("Failed to find the worker to stop ({})", workerName); |
| return; |
| } |
| worker.stop(); |
| workerContainers.remove(workerName); |
| log.info("Worker {} stopped and removed from the map of worker containers", workerName); |
| } |
| |
| public synchronized void stopWorkers() { |
| if (PULSAR_CONTAINERS_LEAVE_RUNNING) { |
| logIgnoringStopDueToLeaveRunning(); |
| return; |
| } |
| // Stop workers that have been initialized |
| workerContainers.values().parallelStream().forEach(WorkerContainer::stop); |
| workerContainers.clear(); |
| } |
| |
| public void startContainers(Map<String, GenericContainer<?>> containers) { |
| containers.forEach((name, container) -> { |
| PulsarContainer.configureLeaveContainerRunning(container); |
| container |
| .withNetwork(network) |
| .withNetworkAliases(name) |
| .start(); |
| log.info("Successfully start container {}.", name); |
| }); |
| } |
| |
| public static void stopContainers(Map<String, GenericContainer<?>> containers) { |
| if (PULSAR_CONTAINERS_LEAVE_RUNNING) { |
| logIgnoringStopDueToLeaveRunning(); |
| return; |
| } |
| containers.values().parallelStream().forEach(GenericContainer::stop); |
| log.info("Successfully stop containers : {}", containers); |
| } |
| |
| private static void logIgnoringStopDueToLeaveRunning() { |
| log.warn("Ignoring stop due to PULSAR_CONTAINERS_LEAVE_RUNNING=true."); |
| } |
| |
| public BrokerContainer getAnyBroker() { |
| return getAnyContainer(brokerContainers, "pulsar-broker"); |
| } |
| |
| public synchronized WorkerContainer getAnyWorker() { |
| return getAnyContainer(workerContainers, "pulsar-functions-worker"); |
| } |
| |
| public synchronized List<WorkerContainer> getAlWorkers() { |
| return new ArrayList<WorkerContainer>(workerContainers.values()); |
| } |
| |
| public BrokerContainer getBroker(int index) { |
| return getAnyContainer(brokerContainers, "pulsar-broker", index); |
| } |
| |
| public synchronized WorkerContainer getWorker(int index) { |
| return getAnyContainer(workerContainers, "pulsar-functions-worker", index); |
| } |
| |
| public synchronized WorkerContainer getWorker(String workerName) { |
| return workerContainers.get(workerName); |
| } |
| |
| private <T> T getAnyContainer(Map<String, T> containers, String serviceName) { |
| List<T> containerList = Lists.newArrayList(); |
| containerList.addAll(containers.values()); |
| Collections.shuffle(containerList); |
| checkArgument(!containerList.isEmpty(), "No " + serviceName + " is alive"); |
| return containerList.get(0); |
| } |
| |
| private <T> T getAnyContainer(Map<String, T> containers, String serviceName, int index) { |
| checkArgument(!containers.isEmpty(), "No " + serviceName + " is alive"); |
| checkArgument((index >= 0 && index < containers.size()), "Index : " + index + " is out range"); |
| return containers.get(serviceName.toLowerCase() + "-" + index); |
| } |
| |
| public Collection<BrokerContainer> getBrokers() { |
| return brokerContainers.values(); |
| } |
| |
| public ProxyContainer getProxy() { |
| return proxyContainer; |
| } |
| |
| public Collection<BKContainer> getBookies() { |
| return bookieContainers.values(); |
| } |
| |
| public ZKContainer getZooKeeper() { |
| return zkContainer; |
| } |
| |
| public ContainerExecResult runAdminCommandOnAnyBroker(String...commands) throws Exception { |
| return runCommandOnAnyBrokerWithScript(ADMIN_SCRIPT, commands); |
| } |
| |
| public ContainerExecResult runPulsarBaseCommandOnAnyBroker(String...commands) throws Exception { |
| return runCommandOnAnyBrokerWithScript(PULSAR_COMMAND_SCRIPT, commands); |
| } |
| |
| private ContainerExecResult runCommandOnAnyBrokerWithScript(String scriptType, String...commands) throws Exception { |
| BrokerContainer container = getAnyBroker(); |
| String[] cmds = new String[commands.length + 1]; |
| cmds[0] = scriptType; |
| System.arraycopy(commands, 0, cmds, 1, commands.length); |
| return container.execCmd(cmds); |
| } |
| |
| public void stopAllBrokers() { |
| brokerContainers.values().forEach(BrokerContainer::stop); |
| } |
| |
| public void startAllBrokers() { |
| brokerContainers.values().forEach(BrokerContainer::start); |
| } |
| |
| public void stopAllBookies() { |
| bookieContainers.values().forEach(BKContainer::stop); |
| } |
| |
| public void startAllBookies() { |
| bookieContainers.values().forEach(BKContainer::start); |
| } |
| |
| public void stopZooKeeper() { |
| zkContainer.stop(); |
| } |
| |
| public void startZooKeeper() { |
| zkContainer.start(); |
| } |
| |
| public ContainerExecResult createNamespace(String nsName) throws Exception { |
| return runAdminCommandOnAnyBroker( |
| "namespaces", "create", "public/" + nsName, |
| "--clusters", clusterName); |
| } |
| |
| public ContainerExecResult createPartitionedTopic(String topicName, int partitions) throws Exception { |
| return runAdminCommandOnAnyBroker( |
| "topics", "create-partitioned-topic", topicName, |
| "-p", String.valueOf(partitions)); |
| } |
| |
| public ContainerExecResult enableDeduplication(String nsName, boolean enabled) throws Exception { |
| return runAdminCommandOnAnyBroker( |
| "namespaces", "set-deduplication", "public/" + nsName, |
| enabled ? "--enable" : "--disable"); |
| } |
| |
| public void dumpFunctionLogs(String name) { |
| for (WorkerContainer container : getAlWorkers()) { |
| log.info("Trying to get function {} logs from container {}", name, container.getContainerName()); |
| try { |
| String logFile = "/pulsar/logs/functions/public/default/" + name + "/" + name + "-0.log"; |
| String logs = container.<String>copyFileFromContainer(logFile, (inputStream) -> { |
| return IOUtils.toString(inputStream, "utf-8"); |
| }); |
| log.info("Function {} logs {}", name, logs); |
| } catch (com.github.dockerjava.api.exception.NotFoundException notFound) { |
| log.info("Cannot download {} logs from {} not found exception {}", name, container.getContainerName(), notFound.toString()); |
| } catch (Throwable err) { |
| log.info("Cannot download {} logs from {}", name, container.getContainerName(), err); |
| } |
| } |
| } |
| |
| private String appendClusterName(String name) { |
| return sharedCsContainer ? clusterName + "-" + name : name; |
| } |
| |
| public BKContainer getAnyBookie() { |
| return getAnyContainer(bookieContainers, "bookie"); |
| } |
| } |
