SerializableZookeeperState implements SerializableState { @SuppressWarnings("unchecked") public <T> T get(String key) throws IOException { try { byte[] existingNodes = zkState.fetch(key).get().value(); if (existingNodes.length > 0) { ByteArrayInputStream bis = new ByteArrayInputStream(existingNodes); ObjectInputStream in = null; try { in = new ObjectInputStream(bis); return (T) in.readObject(); } finally { try { bis.close(); } finally { if (in != null) { in.close(); } } } } else { return null; } } catch (StreamCorruptedException e) { throw new IOException("Corrupted zookeeper zNode. Please delete (rmr) the zNode path using the zookeeper/bin/zkCli.sh tool.", e); } catch (InterruptedException | ClassNotFoundException | ExecutionException | IOException e) { throw new IOException("Unable to get zNode", e); } } SerializableZookeeperState(org.apache.mesos.state.State zkState); @SuppressWarnings("unchecked") T get(String key); void set(String key, T object); void delete(String key);  }
@Test public void testGetValid() throws IOException { String variable = serializableState.get("test"); verify(state, times(1)).fetch(anyString()); assertEquals(SERIALIZABLE_OBJECT, variable); }
@Test public void testNullIfNodeDoesntExist() throws IOException { when(variable.value()).thenReturn(new byte[0]); Object variable = serializableState.get("test"); verify(state, times(1)).fetch(anyString()); assertNull(variable); }
@Test(expected = IOException.class) public void testInterrupted() throws IOException { when(state.fetch(anyString())).thenThrow(InterruptedException.class); serializableState.get("test"); }
@Test(expected = IOException.class) public void testClassNotFound() throws IOException { when(state.fetch(anyString())).thenThrow(ClassNotFoundException.class); serializableState.get("test"); }
@Test(expected = IOException.class) public void testExecution() throws IOException { when(state.fetch(anyString())).thenThrow(ExecutionException.class); serializableState.get("test"); }
@Test(expected = IOException.class) public void testIOException() throws IOException { when(state.fetch(anyString())).thenThrow(IOException.class); serializableState.get("test"); }
@Test(expected = IOException.class) public void testInvalidObjectStream() throws IOException { when(variable.value()).thenReturn("Invalid stream of bytes".getBytes(Charset.forName("UTF-8"))); serializableState.get("test"); }