ARUHA-2324 fixes after merge

src/main/java/org/zalando/nakadi/repository/zookeeper/ZooKeeperHolder.java

@@ -10,40 +10,32 @@
 import org.apache.curator.framework.CuratorFrameworkFactory;
 import org.apache.curator.retry.ExponentialBackoffRetry;
 import org.zalando.nakadi.config.NakadiSettings;
+import org.zalando.nakadi.domain.storage.AddressPort;
+import org.zalando.nakadi.domain.storage.ZookeeperConnection;
 import org.zalando.nakadi.exceptions.runtime.ZookeeperException;
 
 import java.io.Closeable;
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.Collection;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 public class ZooKeeperHolder {
 
     private static final int EXHIBITOR_RETRY_TIME = 1000;
     private static final int EXHIBITOR_RETRY_MAX = 3;
     private static final int EXHIBITOR_POLLING_MS = 300000;
 
-    private final String zookeeperBrokers;
-    private final String zookeeperKafkaNamespace;
-    private final String exhibitorAddresses;
-    private final Integer exhibitorPort;
-    private final Integer sessionTimeoutMs;
     private final Integer connectionTimeoutMs;
     private final long maxCommitTimeoutMs;
+    private final ZookeeperConnection conn;
 
     private CuratorFramework zooKeeper;
     private CuratorFramework subscriptionCurator;
 
-    public ZooKeeperHolder(final ZookeeperConnection zookeeperConnection,
+    public ZooKeeperHolder(final ZookeeperConnection conn,
                            final Integer sessionTimeoutMs,
                            final Integer connectionTimeoutMs,
                            final NakadiSettings nakadiSettings) throws Exception {
-        this.zookeeperBrokers = zookeeperBrokers;
-        this.zookeeperKafkaNamespace = zookeeperKafkaNamespace;
-        this.exhibitorAddresses = exhibitorAddresses;
-        this.exhibitorPort = exhibitorPort;
-        this.sessionTimeoutMs = sessionTimeoutMs;
+        this.conn = conn;
         this.connectionTimeoutMs = connectionTimeoutMs;
         this.maxCommitTimeoutMs = TimeUnit.SECONDS.toMillis(nakadiSettings.getMaxCommitTimeout());
 
@@ -89,7 +81,7 @@ public StaticCuratorFramework(final CuratorFramework curatorFramework) {
         }
 
         @Override
-        public void close() throws IOException {
+        public void close() {
             // do not ever close this particular instance of curator
         }
     }
@@ -101,7 +93,7 @@ public DisposableCuratorFramework(final CuratorFramework curatorFramework) {
         }
 
         @Override
-        public void close() throws IOException {
+        public void close() {
             getCuratorFramework().close();
         }
     }
@@ -119,31 +111,35 @@ private CuratorFramework createCuratorFramework(final int sessionTimeoutMs,
     }
 
     private EnsembleProvider createEnsembleProvider() throws Exception {
-        final EnsembleProvider ensembleProvider;
-        final RetryPolicy retryPolicy = new ExponentialBackoffRetry(EXHIBITOR_RETRY_TIME, EXHIBITOR_RETRY_MAX);
-        if (exhibitorAddresses != null) {
-            final Collection<String> exhibitorHosts = Arrays.asList(exhibitorAddresses.split("\\s*,\\s*"));
-            final Exhibitors exhibitors = new Exhibitors(exhibitorHosts, exhibitorPort,
-                    () -> zookeeperBrokers + zookeeperKafkaNamespace);
-            final ExhibitorRestClient exhibitorRestClient = new DefaultExhibitorRestClient();
-            ensembleProvider = new ExhibitorEnsembleProvider(exhibitors,
-                    exhibitorRestClient, "/exhibitor/v1/cluster/list", EXHIBITOR_POLLING_MS, retryPolicy);
-            ((ExhibitorEnsembleProvider) ensembleProvider).pollForInitialEnsemble();
-        } else {
-            ensembleProvider = new FixedEnsembleProvider(zookeeperBrokers + zookeeperKafkaNamespace);
+        switch (conn.getType()) {
+            case EXHIBITOR:
+                final Exhibitors exhibitors = new Exhibitors(
+                        conn.getAddresses().stream().map(AddressPort::getAddress).collect(Collectors.toList()),
+                        conn.getAddresses().get(0).getPort(),
+                        () -> {
+                            throw new RuntimeException("There is no backup connection string (or it is wrong)");
+                        });
+                final ExhibitorRestClient exhibitorRestClient = new DefaultExhibitorRestClient();
+                final ExhibitorEnsembleProvider result = new ExhibitorEnsembleProvider(
+                        exhibitors,
+                        exhibitorRestClient,
+                        "/exhibitor/v1/cluster/list",
+                        EXHIBITOR_POLLING_MS,
+                        new ExponentialBackoffRetry(EXHIBITOR_RETRY_TIME, EXHIBITOR_RETRY_MAX)) {
+                    @Override
+                    public String getConnectionString() {
+                        return super.getConnectionString() + conn.getPathPrepared();
+                    }
+                };
+                result.pollForInitialEnsemble();
+                return result;
+            case ZOOKEEPER:
+                final String address = conn.getAddresses().stream()
+                        .map(AddressPort::asAddressPort)
+                        .collect(Collectors.joining(","));
+                return new FixedEnsembleProvider(address + conn.getPathPrepared());
+            default:
+                throw new RuntimeException("Connection type " + conn.getType() + " is not supported");
         }
-        return ensembleProvider;
-    }
-
-    private class ExhibitorEnsembleProvider extends org.apache.curator.ensemble.exhibitor.ExhibitorEnsembleProvider {
-
-        ExhibitorEnsembleProvider(final Exhibitors exhibitors, final ExhibitorRestClient restClient,
-                                  final String restUriPath, final int pollingMs, final RetryPolicy retryPolicy) {
-            super(exhibitors, restClient, restUriPath, pollingMs, retryPolicy);
-        }
-    }
-
-    public CuratorFramework get() {
-        return zooKeeper;
     }
 }

src/main/java/org/zalando/nakadi/repository/zookeeper/ZookeeperConfig.java

@@ -5,6 +5,7 @@
 import org.springframework.context.annotation.Profile;
 import org.springframework.core.env.Environment;
 import org.zalando.nakadi.config.NakadiSettings;
+import org.zalando.nakadi.domain.storage.ZookeeperConnection;
 
 @Configuration
 @Profile("!test")