INT-4464 Check ZKMetadataStore.running before use

JIRA: https://jira.spring.io/browse/INT-4464

A `ZookeeperMetadataStore.get()` is based on the `this.cache` variable.
This one is initialized in the `start()`.

* Assert `isRunning()` in the `get()` before using.

**Cherry-pick to 5.0.x and 4.3.x**

Author: artembilan

spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStore.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2016 the original author or authors.
+ * Copyright 2015-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -72,7 +72,7 @@ public class ZookeeperMetadataStore implements ListenableMetadataStore, SmartLif
 
 	private volatile int phase = Integer.MAX_VALUE;
 
-	public ZookeeperMetadataStore(CuratorFramework client) throws Exception {
+	public ZookeeperMetadataStore(CuratorFramework client) {
 		Assert.notNull(client, "Client cannot be null");
 		this.client = client;
 	}
@@ -203,6 +203,7 @@ public void put(String key, String value) {
 	@Override
 	public String get(String key) {
 		Assert.notNull(key, "'key' must not be null.");
+		Assert.state(isRunning(), "ZookeeperMetadataStore has to be started before using.");
 		synchronized (this.updateMap) {
 			ChildData currentData = this.cache.getCurrentData(getPath(key));
 			if (currentData == null) {

spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStoreTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2017 the original author or authors.
+ * Copyright 2015-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -131,7 +131,10 @@ public String evaluate() {
 				return metadataStore.get(testKey2);
 			}
 		})));
+
+		otherMetadataStore.stop();
 		CloseableUtils.closeQuietly(otherClient);
+
 	}
 
 	@Test
@@ -164,6 +167,8 @@ public String evaluate() {
 			}
 		})));
 		assertEquals("Integration-2", otherMetadataStore.get(testKey));
+
+		otherMetadataStore.stop();
 		CloseableUtils.closeQuietly(otherClient);
 	}
 
@@ -223,7 +228,6 @@ public void testRemoveFromMetadataStore() throws Exception {
 		String testValue = "Integration";
 		metadataStore.put(testKey, testValue);
 		assertEquals(testValue, metadataStore.remove(testKey));
-		Thread.sleep(1000);
 		assertNull(metadataStore.remove(testKey));
 	}
 
@@ -296,10 +300,6 @@ public void onUpdate(String key, String newValue) {
 		waitAtBarrier("remove", barriers);
 		assertThat(notifiedChanges, hasSize(4));
 		assertThat(notifiedChanges.get(3), IsIterableContainingInOrder.contains("remove", testKey, "Integration-3"));
-
-		// sleep and try to see if there were any other updates
-		Thread.sleep(1000);
-		assertThat(notifiedChanges, hasSize(4));
 	}
 
 	@Test
@@ -367,9 +367,8 @@ public void onUpdate(String key, String newValue) {
 		assertThat(notifiedChanges, hasSize(4));
 		assertThat(notifiedChanges.get(3), IsIterableContainingInOrder.contains("remove", testKey, "Integration-3"));
 
-		// sleep and try to see if there were any other updates - if there any pending updates, we should catch them by now
-		Thread.sleep(1000);
-		assertThat(notifiedChanges, hasSize(4));
+		otherMetadataStore.stop();
+		CloseableUtils.closeQuietly(otherClient);
 	}
 
 	@Test
@@ -388,6 +387,19 @@ public void testAddRemoveListener() throws Exception {
 		assertThat(listeners, hasSize(0));
 	}
 
+	@Test
+	public void testEnsureStarted() {
+		ZookeeperMetadataStore zookeeperMetadataStore = new ZookeeperMetadataStore(this.client);
+
+		try {
+			zookeeperMetadataStore.get("foo");
+		}
+		catch (Exception e) {
+			assertThat(e, instanceOf(IllegalStateException.class));
+			assertThat(e.getMessage(), containsString("ZookeeperMetadataStore has to be started before using."));
+		}
+	}
+
 	private void waitAtBarrier(String barrierName, Map<String, CyclicBarrier> barriers) {
 		try {
 			barriers.get(barrierName).await(10, TimeUnit.SECONDS);