Skip to content

Commit

Permalink
[improve][cli] Using separate TLS config on the compactor (apache#17426)
Browse files Browse the repository at this point in the history 
Signed-off-by: Zixuan Liu <[email protected]>

Signed-off-by: Zixuan Liu <[email protected]>

### Motivation

Improve the compactor tool, using separate TLS config

### Modifications

- Add separate TLS config on the compactor, both Keystore and PEM formats are supported
- Fix correct use of service URL by `brokerConfig.isBrokerClientTlsEnabled()` value

### Verifying this change

Test has been added.
 
### Documentation

Check the box below or label this PR directly.

Need to update docs? 

- [ ] `doc-required` 
(Your PR needs to update docs and you will update later)
  
- [x] `doc-not-needed` 
(Please explain why)
  
- [ ] `doc` 
(Your PR contains doc changes)

- [ ] `doc-complete`
(Docs have been already added)
  • Loading branch information
@nodece
nodece authored Sep 29, 2022
1 parent 048ccae commit 537aa54
Show file tree
Hide file tree
Showing 2 changed files with 91 additions and 32 deletions.
73 changes: 41 additions & 32 deletions pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactorTool.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -31,13 +31,13 @@
import java.util.concurrent.ScheduledExecutorService;
import lombok.Cleanup;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.common.util.OrderedScheduler;
import org.apache.pulsar.broker.BookKeeperClientFactory;
import org.apache.pulsar.broker.BookKeeperClientFactoryImpl;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.ServiceConfigurationUtils;
import org.apache.pulsar.client.api.ClientBuilder;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.SizeUnit;
import org.apache.pulsar.client.internal.PropertiesUtils;
import org.apache.pulsar.common.configuration.PulsarConfigurationLoader;
Expand Down Expand Up @@ -65,6 +65,45 @@ private static class Arguments {
private boolean generateDocs = false;
}

public static PulsarClient createClient(ServiceConfiguration brokerConfig) throws PulsarClientException {
ClientBuilder clientBuilder = PulsarClient.builder()
.memoryLimit(0, SizeUnit.BYTES);

// Apply all arbitrary configuration. This must be called before setting any fields annotated as
// @Secret on the ClientConfigurationData object because of the way they are serialized.
// See https://github.com/apache/pulsar/issues/8509 for more information.
clientBuilder.loadConf(PropertiesUtils.filterAndMapProperties(brokerConfig.getProperties(), "brokerClient_"));

if (isNotBlank(brokerConfig.getBrokerClientAuthenticationPlugin())) {
clientBuilder.authentication(brokerConfig.getBrokerClientAuthenticationPlugin(),
brokerConfig.getBrokerClientAuthenticationParameters());
}

AdvertisedListener internalListener = ServiceConfigurationUtils.getInternalListener(brokerConfig, "pulsar+ssl");
if (internalListener.getBrokerServiceUrlTls() != null && brokerConfig.isBrokerClientTlsEnabled()) {
clientBuilder.serviceUrl(internalListener.getBrokerServiceUrlTls().toString())
.allowTlsInsecureConnection(brokerConfig.isTlsAllowInsecureConnection());
if (brokerConfig.isBrokerClientTlsEnabledWithKeyStore()) {
clientBuilder.useKeyStoreTls(true)
.tlsKeyStoreType(brokerConfig.getBrokerClientTlsKeyStoreType())
.tlsKeyStorePath(brokerConfig.getBrokerClientTlsKeyStore())
.tlsKeyStorePassword(brokerConfig.getBrokerClientTlsKeyStorePassword())
.tlsTrustStoreType(brokerConfig.getBrokerClientTlsTrustStoreType())
.tlsTrustStorePath(brokerConfig.getBrokerClientTlsTrustStore())
.tlsTrustStorePassword(brokerConfig.getBrokerClientTlsTrustStorePassword());
} else {
clientBuilder.tlsTrustCertsFilePath(brokerConfig.getBrokerClientTrustCertsFilePath())
.tlsKeyFilePath(brokerConfig.getBrokerClientKeyFilePath())
.tlsCertificateFilePath(brokerConfig.getBrokerClientCertificateFilePath());
}
} else {
internalListener = ServiceConfigurationUtils.getInternalListener(brokerConfig, "pulsar");
clientBuilder.serviceUrl(internalListener.getBrokerServiceUrl().toString());
}

return clientBuilder.build();
}

public static void main(String[] args) throws Exception {
Arguments arguments = new Arguments();
JCommander jcommander = new JCommander(arguments);
Expand Down Expand Up @@ -105,40 +144,10 @@ public static void main(String[] args) throws Exception {
);
}

ClientBuilder clientBuilder = PulsarClient.builder()
.memoryLimit(0, SizeUnit.BYTES);

// Apply all arbitrary configuration. This must be called before setting any fields annotated as
// @Secret on the ClientConfigurationData object because of the way they are serialized.
// See https://github.com/apache/pulsar/issues/8509 for more information.
clientBuilder.loadConf(PropertiesUtils.filterAndMapProperties(brokerConfig.getProperties(), "brokerClient_"));

if (isNotBlank(brokerConfig.getBrokerClientAuthenticationPlugin())) {
clientBuilder.authentication(brokerConfig.getBrokerClientAuthenticationPlugin(),
brokerConfig.getBrokerClientAuthenticationParameters());
}

AdvertisedListener internalListener = ServiceConfigurationUtils.getInternalListener(brokerConfig, "pulsar+ssl");
if (internalListener.getBrokerServiceUrlTls() != null) {
log.info("Found a TLS-based advertised listener in configuration file. \n"
+ "Will connect pulsar use TLS.");
clientBuilder
.serviceUrl(internalListener.getBrokerServiceUrlTls().toString())
.allowTlsInsecureConnection(brokerConfig.isTlsAllowInsecureConnection())
.tlsTrustCertsFilePath(brokerConfig.getTlsCertificateFilePath());

} else {
internalListener = ServiceConfigurationUtils.getInternalListener(brokerConfig, "pulsar");
clientBuilder.serviceUrl(internalListener.getBrokerServiceUrl().toString());
}

@Cleanup(value = "shutdownNow")
ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(
new ThreadFactoryBuilder().setNameFormat("compaction-%d").setDaemon(true).build());

@Cleanup(value = "shutdownNow")
OrderedScheduler executor = OrderedScheduler.newSchedulerBuilder().build();

@Cleanup
MetadataStoreExtended store = MetadataStoreExtended.create(brokerConfig.getMetadataStoreUrl(),
MetadataStoreConfig.builder()
Expand All @@ -157,7 +166,7 @@ public static void main(String[] args) throws Exception {
BookKeeper bk = bkClientFactory.create(brokerConfig, store, eventLoopGroup, Optional.empty(), null);

@Cleanup
PulsarClient pulsar = clientBuilder.build();
PulsarClient pulsar = createClient(brokerConfig);

Compactor compactor = new TwoPhaseCompactor(brokerConfig, pulsar, bk, scheduler);
long ledgerId = compactor.compact(arguments.topic).get();
Expand Down
50 changes: 50 additions & 0 deletions pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactorToolTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,13 +18,23 @@
*/
package org.apache.pulsar.compaction;

import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.testng.Assert.assertTrue;
import com.beust.jcommander.Parameter;
import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
import java.lang.reflect.Constructor;
import java.lang.reflect.Field;
import java.util.Arrays;
import java.util.Optional;
import java.util.Properties;
import lombok.Cleanup;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.common.util.CmdGenerateDocs;
import org.testng.annotations.Test;

Expand Down Expand Up @@ -72,4 +82,44 @@ public void testGenerateDocs() throws Exception {
System.setOut(oldStream);
}
}

@Test
public void testUseTlsUrlWithPEM() throws PulsarClientException {
ServiceConfiguration serviceConfiguration = spy(ServiceConfiguration.class);
serviceConfiguration.setBrokerServicePortTls(Optional.of(6651));
serviceConfiguration.setBrokerClientTlsEnabled(true);
serviceConfiguration.setProperties(new Properties());

@Cleanup
PulsarClient ignored = CompactorTool.createClient(serviceConfiguration);

verify(serviceConfiguration, times(1)).isBrokerClientTlsEnabled();
verify(serviceConfiguration, times(1)).isTlsAllowInsecureConnection();
verify(serviceConfiguration, times(1)).getBrokerClientKeyFilePath();
verify(serviceConfiguration, times(1)).getBrokerClientTrustCertsFilePath();
verify(serviceConfiguration, times(1)).getBrokerClientCertificateFilePath();
}

@Test
public void testUseTlsUrlWithKeystore() throws PulsarClientException {
ServiceConfiguration serviceConfiguration = spy(ServiceConfiguration.class);
serviceConfiguration.setBrokerServicePortTls(Optional.of(6651));
serviceConfiguration.setBrokerClientTlsEnabled(true);
serviceConfiguration.setBrokerClientTlsEnabledWithKeyStore(true);
serviceConfiguration.setBrokerClientTlsTrustStore(MockedPulsarServiceBaseTest.BROKER_KEYSTORE_FILE_PATH);

serviceConfiguration.setProperties(new Properties());

@Cleanup
PulsarClient ignored = CompactorTool.createClient(serviceConfiguration);

verify(serviceConfiguration, times(1)).isBrokerClientTlsEnabled();
verify(serviceConfiguration, times(1)).isBrokerClientTlsEnabledWithKeyStore();
verify(serviceConfiguration, times(1)).getBrokerClientTlsKeyStore();
verify(serviceConfiguration, times(1)).getBrokerClientTlsKeyStorePassword();
verify(serviceConfiguration, times(1)).getBrokerClientTlsKeyStoreType();
verify(serviceConfiguration, times(1)).getBrokerClientTlsTrustStore();
verify(serviceConfiguration, times(1)).getBrokerClientTlsTrustStorePassword();
verify(serviceConfiguration, times(1)).getBrokerClientTlsTrustStoreType();
}
}

0 comments on commit 537aa54

Please sign in to comment.