[fix][broker] Fix etcd cluster error and add test for etcd cluster (a…

…pache#16309)
teaey · Aug 2, 2022 · 3d7c112 · 3d7c112
1 parent 7694803
commit 3d7c112
Show file tree

Hide file tree

Showing 2 changed files with 60 additions and 1 deletion.
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/EtcdMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/EtcdMetadataStore.java
@@ -126,7 +126,8 @@ public EtcdMetadataStore(String metadataURL, MetadataStoreConfig conf, boolean e
 
     private Client newEtcdClient(String metadataURL, MetadataStoreConfig conf) throws IOException {
         String etcdUrl = metadataURL.replaceFirst(ETCD_SCHEME_IDENTIFIER, "");
-        ClientBuilder clientBuilder = Client.builder().endpoints(etcdUrl);
+        ClientBuilder clientBuilder = Client.builder()
+                .endpoints(etcdUrl.split(","));
 
         if (StringUtils.isNotEmpty(conf.getConfigFilePath())) {
             try (InputStream inputStream = Files.newInputStream(Paths.get(conf.getConfigFilePath()))) {

diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/impl/EtcdMetadataStoreTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/impl/EtcdMetadataStoreTest.java
@@ -24,6 +24,7 @@
 import com.google.common.io.Resources;
 import io.etcd.jetcd.launcher.EtcdCluster;
 import io.etcd.jetcd.launcher.EtcdClusterFactory;
+
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
@@ -40,6 +41,63 @@
 @Slf4j
 public class EtcdMetadataStoreTest {
 
+    @Test
+    public void testCluster() throws Exception {
+        @Cleanup
+        EtcdCluster etcdCluster = EtcdClusterFactory.buildCluster("test-cluster", 3, false);
+        etcdCluster.start();
+
+        EtcdConfig etcdConfig = EtcdConfig.builder().useTls(false)
+                .tlsProvider(null)
+                .authority("etcd0")
+                .build();
+
+        Path etcdConfigPath = Files.createTempFile("etcd_config_cluster", ".yml");
+        new ObjectMapper(new YAMLFactory()).writeValue(etcdConfigPath.toFile(), etcdConfig);
+
+        String metadataURL =
+                "etcd:" + etcdCluster.getClientEndpoints().stream().map(URI::toString).collect(Collectors.joining(","));
+
+        @Cleanup
+        MetadataStore store = MetadataStoreFactory.create(metadataURL,
+                MetadataStoreConfig.builder().configFilePath(etcdConfigPath.toString()).build());
+
+        store.put("/test", "value".getBytes(StandardCharsets.UTF_8), Optional.empty()).join();
+
+        assertTrue(store.exists("/test").join());
+
+    }
+
+    @Test
+    public void testClusterWithTls() throws Exception {
+        @Cleanup
+        EtcdCluster etcdCluster = EtcdClusterFactory.buildCluster("test-cluster", 3, true);
+        etcdCluster.start();
+
+        EtcdConfig etcdConfig = EtcdConfig.builder().useTls(true)
+                .tlsProvider(null)
+                .authority("etcd0")
+                .tlsTrustCertsFilePath(Resources.getResource("ssl/cert/ca.pem").getPath())
+                .tlsKeyFilePath(Resources.getResource("ssl/cert/client-key-pk8.pem").getPath())
+                .tlsCertificateFilePath(Resources.getResource("ssl/cert/client.pem").getPath())
+                .build();
+
+        Path etcdConfigPath = Files.createTempFile("etcd_config_cluster_ssl", ".yml");
+        new ObjectMapper(new YAMLFactory()).writeValue(etcdConfigPath.toFile(), etcdConfig);
+
+        String metadataURL =
+                "etcd:" + etcdCluster.getClientEndpoints().stream().map(URI::toString).collect(Collectors.joining(","));
+
+        @Cleanup
+        MetadataStore store = MetadataStoreFactory.create(metadataURL,
+                MetadataStoreConfig.builder().configFilePath(etcdConfigPath.toString()).build());
+
+        store.put("/test", "value".getBytes(StandardCharsets.UTF_8), Optional.empty()).join();
+
+        assertTrue(store.exists("/test").join());
+
+    }
+
     @Test
     public void testTlsInstance() throws Exception {
         @Cleanup