@Test public void testClusterListenerGetsNotifiedOfUpdate() { long time = 0; MockClusterResourceListener mockClusterListener = new MockClusterResourceListener(); ClusterResourceListeners listeners = new ClusterResourceListeners(); listeners.maybeAdd(mockClusterListener); metadata = new Metadata(refreshBackoffMs, metadataExpireMs, true, false, listeners); String hostName = "www.example.com"; Cluster cluster = Cluster.bootstrap(Arrays.asList(new InetSocketAddress(hostName, 9002))); metadata.update(cluster, Collections.<String>emptySet(), time); assertFalse("ClusterResourceListener should not called when metadata is updated with bootstrap Cluster", MockClusterResourceListener.IS_ON_UPDATE_CALLED.get()); metadata.update(new Cluster( "dummy", Arrays.asList(new Node(0, "host1", 1000)), Arrays.asList( new PartitionInfo("topic", 0, null, null, null), new PartitionInfo("topic1", 0, null, null, null)), Collections.<String>emptySet(), Collections.<String>emptySet()), Collections.<String>emptySet(), 100); assertEquals("MockClusterResourceListener did not get cluster metadata correctly", "dummy", mockClusterListener.clusterResource().clusterId()); assertTrue("MockClusterResourceListener should be called when metadata is updated with non-bootstrap Cluster", MockClusterResourceListener.IS_ON_UPDATE_CALLED.get()); }
public synchronized void update(Cluster cluster, Set<String> unavailableTopics, long now) { Objects.requireNonNull(cluster, "cluster should not be null"); this.needUpdate = false; this.lastRefreshMs = now; this.lastSuccessfulRefreshMs = now; this.version += 1; if (topicExpiryEnabled) { for (Iterator<Map.Entry<String, Long>> it = topics.entrySet().iterator(); it.hasNext(); ) { Map.Entry<String, Long> entry = it.next(); long expireMs = entry.getValue(); if (expireMs == TOPIC_EXPIRY_NEEDS_UPDATE) entry.setValue(now + TOPIC_EXPIRY_MS); else if (expireMs <= now) { it.remove(); log.debug("Removing unused topic {} from the metadata list, expiryMs {} now {}", entry.getKey(), expireMs, now); } } } for (Listener listener: listeners) listener.onMetadataUpdate(cluster, unavailableTopics); String previousClusterId = cluster.clusterResource().clusterId(); if (this.needMetadataForAllTopics) { this.needUpdate = false; this.cluster = getClusterForCurrentTopics(cluster); } else { this.cluster = cluster; } if (!cluster.isBootstrapConfigured()) { String clusterId = cluster.clusterResource().clusterId(); if (clusterId == null ? previousClusterId != null : !clusterId.equals(previousClusterId)) log.info("Cluster ID: {}", cluster.clusterResource().clusterId()); clusterResourceListeners.onUpdate(cluster.clusterResource()); } notifyAll(); log.debug("Updated cluster metadata version {} to {}", this.version, this.cluster); }
Metadata { public synchronized void update(Cluster cluster, Set<String> unavailableTopics, long now) { Objects.requireNonNull(cluster, "cluster should not be null"); this.needUpdate = false; this.lastRefreshMs = now; this.lastSuccessfulRefreshMs = now; this.version += 1; if (topicExpiryEnabled) { for (Iterator<Map.Entry<String, Long>> it = topics.entrySet().iterator(); it.hasNext(); ) { Map.Entry<String, Long> entry = it.next(); long expireMs = entry.getValue(); if (expireMs == TOPIC_EXPIRY_NEEDS_UPDATE) entry.setValue(now + TOPIC_EXPIRY_MS); else if (expireMs <= now) { it.remove(); log.debug("Removing unused topic {} from the metadata list, expiryMs {} now {}", entry.getKey(), expireMs, now); } } } for (Listener listener: listeners) listener.onMetadataUpdate(cluster, unavailableTopics); String previousClusterId = cluster.clusterResource().clusterId(); if (this.needMetadataForAllTopics) { this.needUpdate = false; this.cluster = getClusterForCurrentTopics(cluster); } else { this.cluster = cluster; } if (!cluster.isBootstrapConfigured()) { String clusterId = cluster.clusterResource().clusterId(); if (clusterId == null ? previousClusterId != null : !clusterId.equals(previousClusterId)) log.info("Cluster ID: {}", cluster.clusterResource().clusterId()); clusterResourceListeners.onUpdate(cluster.clusterResource()); } notifyAll(); log.debug("Updated cluster metadata version {} to {}", this.version, this.cluster); } }
Metadata { public synchronized void update(Cluster cluster, Set<String> unavailableTopics, long now) { Objects.requireNonNull(cluster, "cluster should not be null"); this.needUpdate = false; this.lastRefreshMs = now; this.lastSuccessfulRefreshMs = now; this.version += 1; if (topicExpiryEnabled) { for (Iterator<Map.Entry<String, Long>> it = topics.entrySet().iterator(); it.hasNext(); ) { Map.Entry<String, Long> entry = it.next(); long expireMs = entry.getValue(); if (expireMs == TOPIC_EXPIRY_NEEDS_UPDATE) entry.setValue(now + TOPIC_EXPIRY_MS); else if (expireMs <= now) { it.remove(); log.debug("Removing unused topic {} from the metadata list, expiryMs {} now {}", entry.getKey(), expireMs, now); } } } for (Listener listener: listeners) listener.onMetadataUpdate(cluster, unavailableTopics); String previousClusterId = cluster.clusterResource().clusterId(); if (this.needMetadataForAllTopics) { this.needUpdate = false; this.cluster = getClusterForCurrentTopics(cluster); } else { this.cluster = cluster; } if (!cluster.isBootstrapConfigured()) { String clusterId = cluster.clusterResource().clusterId(); if (clusterId == null ? previousClusterId != null : !clusterId.equals(previousClusterId)) log.info("Cluster ID: {}", cluster.clusterResource().clusterId()); clusterResourceListeners.onUpdate(cluster.clusterResource()); } notifyAll(); log.debug("Updated cluster metadata version {} to {}", this.version, this.cluster); } Metadata(long refreshBackoffMs, long metadataExpireMs, boolean allowAutoTopicCreation);  Metadata(long refreshBackoffMs, long metadataExpireMs, boolean allowAutoTopicCreation,
                    boolean topicExpiryEnabled, ClusterResourceListeners clusterResourceListeners); }
Metadata { public synchronized void update(Cluster cluster, Set<String> unavailableTopics, long now) { Objects.requireNonNull(cluster, "cluster should not be null"); this.needUpdate = false; this.lastRefreshMs = now; this.lastSuccessfulRefreshMs = now; this.version += 1; if (topicExpiryEnabled) { for (Iterator<Map.Entry<String, Long>> it = topics.entrySet().iterator(); it.hasNext(); ) { Map.Entry<String, Long> entry = it.next(); long expireMs = entry.getValue(); if (expireMs == TOPIC_EXPIRY_NEEDS_UPDATE) entry.setValue(now + TOPIC_EXPIRY_MS); else if (expireMs <= now) { it.remove(); log.debug("Removing unused topic {} from the metadata list, expiryMs {} now {}", entry.getKey(), expireMs, now); } } } for (Listener listener: listeners) listener.onMetadataUpdate(cluster, unavailableTopics); String previousClusterId = cluster.clusterResource().clusterId(); if (this.needMetadataForAllTopics) { this.needUpdate = false; this.cluster = getClusterForCurrentTopics(cluster); } else { this.cluster = cluster; } if (!cluster.isBootstrapConfigured()) { String clusterId = cluster.clusterResource().clusterId(); if (clusterId == null ? previousClusterId != null : !clusterId.equals(previousClusterId)) log.info("Cluster ID: {}", cluster.clusterResource().clusterId()); clusterResourceListeners.onUpdate(cluster.clusterResource()); } notifyAll(); log.debug("Updated cluster metadata version {} to {}", this.version, this.cluster); } Metadata(long refreshBackoffMs, long metadataExpireMs, boolean allowAutoTopicCreation);  Metadata(long refreshBackoffMs, long metadataExpireMs, boolean allowAutoTopicCreation,
                    boolean topicExpiryEnabled, ClusterResourceListeners clusterResourceListeners); synchronized Cluster fetch(); synchronized void add(String topic); synchronized long timeToNextUpdate(long nowMs); synchronized int requestUpdate(); synchronized boolean updateRequested(); synchronized void awaitUpdate(final int lastVersion, final long maxWaitMs); synchronized void setTopics(Collection<String> topics); synchronized Set<String> topics(); synchronized boolean containsTopic(String topic); synchronized void update(Cluster cluster, Set<String> unavailableTopics, long now); synchronized void failedUpdate(long now); synchronized int version(); synchronized long lastSuccessfulUpdate(); boolean allowAutoTopicCreation(); synchronized void needMetadataForAllTopics(boolean needMetadataForAllTopics); synchronized boolean needMetadataForAllTopics(); synchronized void addListener(Listener listener); synchronized void removeListener(Listener listener); }
Metadata { public synchronized void update(Cluster cluster, Set<String> unavailableTopics, long now) { Objects.requireNonNull(cluster, "cluster should not be null"); this.needUpdate = false; this.lastRefreshMs = now; this.lastSuccessfulRefreshMs = now; this.version += 1; if (topicExpiryEnabled) { for (Iterator<Map.Entry<String, Long>> it = topics.entrySet().iterator(); it.hasNext(); ) { Map.Entry<String, Long> entry = it.next(); long expireMs = entry.getValue(); if (expireMs == TOPIC_EXPIRY_NEEDS_UPDATE) entry.setValue(now + TOPIC_EXPIRY_MS); else if (expireMs <= now) { it.remove(); log.debug("Removing unused topic {} from the metadata list, expiryMs {} now {}", entry.getKey(), expireMs, now); } } } for (Listener listener: listeners) listener.onMetadataUpdate(cluster, unavailableTopics); String previousClusterId = cluster.clusterResource().clusterId(); if (this.needMetadataForAllTopics) { this.needUpdate = false; this.cluster = getClusterForCurrentTopics(cluster); } else { this.cluster = cluster; } if (!cluster.isBootstrapConfigured()) { String clusterId = cluster.clusterResource().clusterId(); if (clusterId == null ? previousClusterId != null : !clusterId.equals(previousClusterId)) log.info("Cluster ID: {}", cluster.clusterResource().clusterId()); clusterResourceListeners.onUpdate(cluster.clusterResource()); } notifyAll(); log.debug("Updated cluster metadata version {} to {}", this.version, this.cluster); } Metadata(long refreshBackoffMs, long metadataExpireMs, boolean allowAutoTopicCreation);  Metadata(long refreshBackoffMs, long metadataExpireMs, boolean allowAutoTopicCreation,
                    boolean topicExpiryEnabled, ClusterResourceListeners clusterResourceListeners); synchronized Cluster fetch(); synchronized void add(String topic); synchronized long timeToNextUpdate(long nowMs); synchronized int requestUpdate(); synchronized boolean updateRequested(); synchronized void awaitUpdate(final int lastVersion, final long maxWaitMs); synchronized void setTopics(Collection<String> topics); synchronized Set<String> topics(); synchronized boolean containsTopic(String topic); synchronized void update(Cluster cluster, Set<String> unavailableTopics, long now); synchronized void failedUpdate(long now); synchronized int version(); synchronized long lastSuccessfulUpdate(); boolean allowAutoTopicCreation(); synchronized void needMetadataForAllTopics(boolean needMetadataForAllTopics); synchronized boolean needMetadataForAllTopics(); synchronized void addListener(Listener listener); synchronized void removeListener(Listener listener); static final long TOPIC_EXPIRY_MS; }