Merge remote-tracking branch 'origin/master' into ARUHA-1501-choose-p…

…artition-hila # Conflicts: # CHANGELOG.md # src/main/java/org/zalando/nakadi/service/subscription/model/Session.java # src/main/java/org/zalando/nakadi/service/subscription/zk/AbstractZkSubscriptionClient.java
zalando · Mar 26, 2018 · be59718 · be59718
2 parents 43ee61a + 43e8a1f
commit be59718
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 0 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -6,9 +6,16 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ## [Unreleased]
 
+## [2.6.0] - 2018-03-26
+
 ### Added
 - Allow to select partitions to read from a subscription
 
+## [2.5.10] - 2018-03-26
+
+### Added
+- Added support of future format of session in ZK
+
 ## [2.5.9] - 2018-03-06
 
 ### Changed

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
@@ -174,6 +174,17 @@ public Map<EventTypePartition, SubscriptionCursorWithoutToken> getOffsets(
                 new SubscriptionCursorWithoutToken(etp.getEventType(), etp.getPartition(), new String(value, UTF_8)));
     }
 
+    protected Session deserializeSession(final String sessionId, final byte[] sessionZkData) throws IOException {
+        try {
+            // old version of session: zkNode data is session weight
+            final int weight = Integer.parseInt(new String(sessionZkData, UTF_8));
+            return new Session(sessionId, weight);
+        } catch (final NumberFormatException nfe) {
+            // new version of session: zkNode data is session object as json
+            return objectMapper.readValue(sessionZkData, Session.class);
+        }
+    }
+
     @Override
     public void transfer(final String sessionId, final Collection<EventTypePartition> partitions)
             throws NakadiRuntimeException, SubscriptionNotInitializedException {