From 8ea76cd9dd69fede5e55d4fb8b8a40da3d9c3129 Mon Sep 17 00:00:00 2001
From: antban <dmitry.sorokin@gmail.com>
Date: Wed, 31 May 2017 10:11:45 +0200
Subject: [PATCH] ARUHA-834 Fixes after review

---
 .../zk/NewZkSubscriptionClient.java           | 32 ++++++++-----------
 .../zk/OldZkSubscriptionClient.java           |  2 +-
 .../subscription/zk/ZkSubscriptionClient.java |  6 ++--
 3 files changed, 18 insertions(+), 22 deletions(-)

diff --git a/src/main/java/org/zalando/nakadi/service/subscription/zk/NewZkSubscriptionClient.java b/src/main/java/org/zalando/nakadi/service/subscription/zk/NewZkSubscriptionClient.java
index d55072f518..23d833a05a 100644
--- a/src/main/java/org/zalando/nakadi/service/subscription/zk/NewZkSubscriptionClient.java
+++ b/src/main/java/org/zalando/nakadi/service/subscription/zk/NewZkSubscriptionClient.java
@@ -22,7 +22,7 @@
  * Subscription client that uses topology as an object node and separates changeable data to a separate zk node.
  * The structure of zk data is like this:
  * <pre>
- *- nakadi
+ * - nakadi
  * +- locks
  * | |- subscription_{subscription_id}      // Node that is being used to guarantee consistency of subscription data
  * |
@@ -142,10 +142,10 @@ protected byte[] createTopologyAndOffsets(final Collection<SubscriptionCursorWit
     }
 
     @Override
-    public void updatePartitionsConfiguration(final Partition[] partitions) throws NakadiRuntimeException {
+    public void updatePartitionsConfiguration(final Partition[] partitions) throws NakadiRuntimeException,
+            SubscriptionNotInitializedException, OldSubscriptionFormatException {
+        final Topology newTopology = readTopology().withUpdatedPartitions(partitions);
         try {
-            final Topology topology = readTopology();
-            final Topology newTopology = topology.withUpdatedPartitions(partitions);
             getCurator().setData().forPath(
                     getSubscriptionPath(NODE_TOPOLOGY),
                     objectMapper.writeValueAsBytes(newTopology));
@@ -154,7 +154,8 @@ public void updatePartitionsConfiguration(final Partition[] partitions) throws N
         }
     }
 
-    private Topology readTopology() throws Exception {
+    private Topology readTopology() throws NakadiRuntimeException, OldSubscriptionFormatException,
+            SubscriptionNotInitializedException {
         try {
             final byte[] data = getCurator().getData().forPath(getSubscriptionPath(NODE_TOPOLOGY));
             final Topology result = objectMapper.readValue(data, Topology.class);
@@ -164,19 +165,15 @@ private Topology readTopology() throws Exception {
             throw new OldSubscriptionFormatException();
         } catch (KeeperException.NoNodeException ex) {
             throw new SubscriptionNotInitializedException(getSubscriptionId());
+        } catch (final Exception ex) {
+            throw new NakadiRuntimeException(ex);
         }
     }
 
     @Override
     public Partition[] listPartitions() throws NakadiRuntimeException, SubscriptionNotInitializedException,
             OldSubscriptionFormatException {
-        try {
-            return readTopology().getPartitions();
-        } catch (final SubscriptionNotInitializedException | OldSubscriptionFormatException e) {
-            throw e;
-        } catch (final Exception e) {
-            throw new NakadiRuntimeException(e);
-        }
+        return readTopology().getPartitions();
     }
 
     protected String getOffsetPath(final EventTypePartition etp) {
@@ -194,14 +191,11 @@ public SubscriptionCursorWithoutToken getOffset(final EventTypePartition key) th
     }
 
     @Override
-    public void transfer(final String sessionId, final Collection<EventTypePartition> partitions) {
+    public void transfer(final String sessionId, final Collection<EventTypePartition> partitions)
+            throws NakadiRuntimeException, OldSubscriptionFormatException, SubscriptionNotInitializedException {
         getLog().info("session " + sessionId + " releases partitions " + partitions);
-        final Topology topology;
-        try {
-            topology = readTopology();
-        } catch (final Exception e) {
-            throw new NakadiRuntimeException(e);
-        }
+        final Topology topology = readTopology();
+
         final List<Partition> changeSet = new ArrayList<>();
         for (final EventTypePartition etp : partitions) {
             final Partition candidate = Stream.of(topology.getPartitions())
diff --git a/src/main/java/org/zalando/nakadi/service/subscription/zk/OldZkSubscriptionClient.java b/src/main/java/org/zalando/nakadi/service/subscription/zk/OldZkSubscriptionClient.java
index 19e4f484b2..38f9e2f621 100644
--- a/src/main/java/org/zalando/nakadi/service/subscription/zk/OldZkSubscriptionClient.java
+++ b/src/main/java/org/zalando/nakadi/service/subscription/zk/OldZkSubscriptionClient.java
@@ -126,7 +126,7 @@ private void incrementTopology() {
     }
 
     @Override
-    public Partition[] listPartitions() {
+    public Partition[] listPartitions() throws NakadiRuntimeException {
         getLog().info("fetching partitions information");
 
         final List<String> zkPartitions;
diff --git a/src/main/java/org/zalando/nakadi/service/subscription/zk/ZkSubscriptionClient.java b/src/main/java/org/zalando/nakadi/service/subscription/zk/ZkSubscriptionClient.java
index 369adb399d..40aef85bf4 100644
--- a/src/main/java/org/zalando/nakadi/service/subscription/zk/ZkSubscriptionClient.java
+++ b/src/main/java/org/zalando/nakadi/service/subscription/zk/ZkSubscriptionClient.java
@@ -49,7 +49,8 @@ public interface ZkSubscriptionClient {
     /**
      * Updates specified partitions in zk.
      */
-    void updatePartitionsConfiguration(Partition[] partitions) throws NakadiRuntimeException;
+    void updatePartitionsConfiguration(Partition[] partitions) throws NakadiRuntimeException,
+            SubscriptionNotInitializedException, OldSubscriptionFormatException;
 
     /**
      * Returns session list in zk related to this subscription.
@@ -113,7 +114,8 @@ List<Boolean> commitOffsets(List<SubscriptionCursorWithoutToken> cursors,
      * @param sessionId  Someone who actually tries to transfer data.
      * @param partitions topic ids and partition ids of transferred data.
      */
-    void transfer(String sessionId, Collection<EventTypePartition> partitions);
+    void transfer(String sessionId, Collection<EventTypePartition> partitions)
+            throws NakadiRuntimeException, OldSubscriptionFormatException, SubscriptionNotInitializedException;
 
     /**
      * Retrieves subscription data like partitions and sessions from ZK under lock.