Skip to content

Commit

Permalink
[fix][test] AdvertisedListenersTest.setup (#17869)
Browse files Browse the repository at this point in the history
  • Loading branch information
congbobo184 authored Oct 17, 2022
1 parent a990822 commit b530184
Show file tree
Hide file tree
Showing 10 changed files with 158 additions and 19 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
*/
package org.apache.bookkeeper.mledger.impl;

import static org.apache.pulsar.common.util.PortManager.releaseLockedPort;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
Expand Down Expand Up @@ -117,7 +118,7 @@ public void testBookieFailure() throws Exception {
metadataStore.unsetAlwaysFail();

bkc = new BookKeeperTestClient(baseClientConf);
startNewBookie();
int port = startNewBookie();

// Reconnect a new bk client
factory.shutdown();
Expand Down Expand Up @@ -147,6 +148,7 @@ public void testBookieFailure() throws Exception {
assertEquals("entry-2", new String(entries.get(0).getData()));
entries.forEach(Entry::release);
factory.shutdown();
releaseLockedPort(port);
}

@Test
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
package org.apache.bookkeeper.test;

import static org.apache.bookkeeper.util.BookKeeperConstants.AVAILABLE_NODE;
import static org.apache.pulsar.common.util.PortManager.nextLockedFreePort;
import static org.testng.Assert.assertFalse;

import com.google.common.base.Stopwatch;
Expand Down Expand Up @@ -62,7 +63,7 @@
import org.apache.bookkeeper.proto.BookieServer;
import org.apache.bookkeeper.replication.Auditor;
import org.apache.bookkeeper.replication.ReplicationWorker;
import org.apache.bookkeeper.util.PortManager;
import org.apache.pulsar.common.util.PortManager;
import org.apache.pulsar.metadata.api.MetadataStoreConfig;
import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
import org.apache.pulsar.metadata.impl.FaultInjectionMetadataStore;
Expand Down Expand Up @@ -113,6 +114,7 @@ public void handleTestMethodName(Method method) {

private boolean isAutoRecoveryEnabled;
protected ExecutorService executor;
private final List<Integer> bookiePorts = new ArrayList<>();

SynchronousQueue<Throwable> asyncExceptions = new SynchronousQueue<>();
protected void captureThrowable(Runnable c) {
Expand Down Expand Up @@ -264,7 +266,7 @@ protected void startBKCluster(String metadataServiceUri) throws Exception {

// Create Bookie Servers (B1, B2, B3)
for (int i = 0; i < numBookies; i++) {
startNewBookie();
bookiePorts.add(startNewBookie());
}
}

Expand All @@ -283,14 +285,15 @@ protected void stopBKCluster() throws Exception {
t.shutdown();
}
servers.clear();
bookiePorts.removeIf(PortManager::releaseLockedPort);
}

protected ServerConfiguration newServerConfiguration() throws Exception {
File f = tmpDirs.createNew("bookie", "test");

int port;
if (baseConf.isEnableLocalTransport() || !baseConf.getAllowEphemeralPorts()) {
port = PortManager.nextFreePort();
port = nextLockedFreePort();
} else {
port = 0;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -21,12 +21,14 @@
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminBuilder;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.common.util.PortManager;
import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
import org.apache.pulsar.metadata.impl.ZKMetadataStore;
Expand Down Expand Up @@ -124,6 +126,9 @@ protected void additionalBrokersCleanup() {
try {
pulsarService.getConfiguration().setBrokerShutdownTimeoutMs(0L);
pulsarService.close();
pulsarService.getConfiguration().getBrokerServicePort().ifPresent(PortManager::releaseLockedPort);
pulsarService.getConfiguration().getWebServicePort().ifPresent(PortManager::releaseLockedPort);
pulsarService.getConfiguration().getWebServicePortTls().ifPresent(PortManager::releaseLockedPort);
} catch (PulsarServerException e) {
// ignore
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,14 +18,14 @@
*/
package org.apache.pulsar.broker.loadbalance;

import static org.apache.pulsar.common.util.PortManager.nextLockedFreePort;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;

import java.net.URI;
import java.util.Optional;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.util.PortManager;
import org.apache.http.HttpEntity;
import org.apache.http.HttpHeaders;
import org.apache.http.client.methods.CloseableHttpResponse;
Expand Down Expand Up @@ -66,9 +66,9 @@ protected ServiceConfiguration createConfForAdditionalBroker(int additionalBroke
}

private void updateConfig(ServiceConfiguration conf, String advertisedAddress) {
int pulsarPort = PortManager.nextFreePort();
int httpPort = PortManager.nextFreePort();
int httpsPort = PortManager.nextFreePort();
int pulsarPort = nextLockedFreePort();
int httpPort = nextLockedFreePort();
int httpsPort = nextLockedFreePort();

// Use invalid domain name as identifier and instead make sure the advertised listeners work as intended
conf.setAdvertisedAddress(advertisedAddress);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -22,12 +22,12 @@
import io.netty.channel.*;
import io.netty.channel.socket.SocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.util.PortManager;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.IOUtils;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.BrokerTestBase;
import org.apache.pulsar.common.util.PortManager;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
Expand All @@ -39,11 +39,14 @@
import java.net.SocketAddress;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.zip.ZipEntry;
import java.util.zip.ZipOutputStream;

import static org.apache.pulsar.common.util.PortManager.nextLockedFreePort;
import static org.testng.Assert.assertEquals;

@Slf4j
Expand All @@ -54,6 +57,8 @@ public static final class MyProtocolHandler implements ProtocolHandler {

private ServiceConfiguration conf;

private final List<Integer> ports = new ArrayList<>();

@Override
public String protocolName() {
return "test";
Expand Down Expand Up @@ -81,7 +86,9 @@ public void start(BrokerService service) {

@Override
public Map<InetSocketAddress, ChannelInitializer<SocketChannel>> newChannelInitializers() {
return Collections.singletonMap(new InetSocketAddress(conf.getBindAddress(), PortManager.nextFreePort()),
int port = nextLockedFreePort();
this.ports.add(port);
return Collections.singletonMap(new InetSocketAddress(conf.getBindAddress(), port),
new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
Expand All @@ -106,7 +113,7 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {

@Override
public void close() {

ports.removeIf(PortManager::releaseLockedPort);
}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
*/
package org.apache.pulsar.functions.worker;

import static org.apache.pulsar.common.util.PortManager.nextLockedFreePort;
import static org.testng.Assert.assertEquals;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
Expand All @@ -33,7 +34,6 @@
import java.util.Optional;
import java.util.Set;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.util.PortManager;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.authentication.AuthenticationProviderTls;
Expand All @@ -44,6 +44,7 @@
import org.apache.pulsar.common.policies.data.TenantInfo;
import org.apache.pulsar.common.util.ClassLoaderUtils;
import org.apache.pulsar.common.util.ObjectMapperFactory;
import org.apache.pulsar.common.util.PortManager;
import org.apache.pulsar.functions.api.utils.IdentityFunction;
import org.apache.pulsar.functions.runtime.thread.ThreadRuntimeFactory;
import org.apache.pulsar.functions.runtime.thread.ThreadRuntimeFactoryConfig;
Expand Down Expand Up @@ -86,8 +87,8 @@ void setup() throws Exception {

// start brokers
for (int i = 0; i < BROKER_COUNT; i++) {
int brokerPort = PortManager.nextFreePort();
int webPort = PortManager.nextFreePort();
int brokerPort = nextLockedFreePort();
int webPort = nextLockedFreePort();

ServiceConfiguration config = new ServiceConfiguration();
config.setBrokerShutdownTimeoutMs(0L);
Expand Down Expand Up @@ -196,6 +197,10 @@ void tearDown() throws Exception {
for (int i = 0; i < BROKER_COUNT; i++) {
if (pulsarServices[i] != null) {
pulsarServices[i].close();
pulsarServices[i].getConfiguration().
getBrokerServicePort().ifPresent(PortManager::releaseLockedPort);
pulsarServices[i].getConfiguration()
.getWebServicePort().ifPresent(PortManager::releaseLockedPort);
}
}
bkEnsemble.stop();
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,69 @@
/**
* 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.common.util;

import java.net.ServerSocket;
import java.util.HashSet;
import java.util.Set;

public class PortManager {

private static final Set<Integer> PORTS = new HashSet<>();

/**
* Return a locked available port.
*
* @return locked available port.
*/
public static synchronized int nextLockedFreePort() {
int exceptionCount = 0;
while (true) {
try (ServerSocket ss = new ServerSocket(0)) {
int port = ss.getLocalPort();
if (!checkPortIfLocked(port)) {
PORTS.add(port);
return port;
}
} catch (Exception e) {
exceptionCount++;
if (exceptionCount > 100) {
throw new RuntimeException("Unable to allocate socket port", e);
}
}
}
}

/**
* Returns whether the port was released successfully.
*
* @return whether the release is successful.
*/
public static synchronized boolean releaseLockedPort(int lockedPort) {
return PORTS.remove(lockedPort);
}

/**
* Check port if locked.
*
* @return whether the port is locked.
*/
public static synchronized boolean checkPortIfLocked(int lockedPort) {
return PORTS.contains(lockedPort);
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,37 @@
/**
* 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.common.util;

import org.testng.annotations.Test;

import static org.apache.pulsar.common.util.PortManager.checkPortIfLocked;
import static org.apache.pulsar.common.util.PortManager.nextLockedFreePort;
import static org.apache.pulsar.common.util.PortManager.releaseLockedPort;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;

public class PortManagerTest {
@Test
public void testCheckPortIfLockedAndRemove() {
int port = nextLockedFreePort();
assertTrue(checkPortIfLocked(port));
assertTrue(releaseLockedPort(port));
assertFalse(checkPortIfLocked(port));
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
import static org.apache.bookkeeper.bookie.BookKeeperServerStats.LD_INDEX_SCOPE;
import static org.apache.bookkeeper.bookie.BookKeeperServerStats.LD_LEDGER_SCOPE;
import static org.apache.bookkeeper.util.BookKeeperConstants.AVAILABLE_NODE;
import static org.apache.pulsar.common.util.PortManager.nextLockedFreePort;
import static org.testng.Assert.assertFalse;
import com.google.common.base.Stopwatch;
import java.io.File;
Expand Down Expand Up @@ -83,7 +84,7 @@
import org.apache.bookkeeper.test.ZooKeeperCluster;
import org.apache.bookkeeper.test.ZooKeeperClusterUtil;
import org.apache.bookkeeper.util.DiskChecker;
import org.apache.bookkeeper.util.PortManager;
import org.apache.pulsar.common.util.PortManager;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
Expand Down Expand Up @@ -129,6 +130,8 @@ public void handleTestMethodName(Method method) {

private boolean isAutoRecoveryEnabled;

private final List<Integer> bookiePorts = new ArrayList<>();

SynchronousQueue<Throwable> asyncExceptions = new SynchronousQueue<>();
protected void captureThrowable(Runnable c) {
try {
Expand Down Expand Up @@ -282,14 +285,15 @@ protected void stopBKCluster() throws Exception {
t.shutdown();
}
servers.clear();
bookiePorts.removeIf(PortManager::releaseLockedPort);
}

protected ServerConfiguration newServerConfiguration() throws Exception {
File f = tmpDirs.createNew("bookie", "test");

int port;
if (baseConf.isEnableLocalTransport() || !baseConf.getAllowEphemeralPorts()) {
port = PortManager.nextFreePort();
port = nextLockedFreePort();
} else {
port = 0;
}
Expand Down
Loading

0 comments on commit b530184

Please sign in to comment.