@Test public void testMetadata() throws Exception { long time = 0; metadata.update(Cluster.empty(), Collections.<String>emptySet(), time); assertFalse("No update needed.", metadata.timeToNextUpdate(time) == 0); metadata.requestUpdate(); assertFalse("Still no updated needed due to backoff", metadata.timeToNextUpdate(time) == 0); time += refreshBackoffMs; assertTrue("Update needed now that backoff time expired", metadata.timeToNextUpdate(time) == 0); String topic = "my-topic"; Thread t1 = asyncFetch(topic, 500); Thread t2 = asyncFetch(topic, 500); assertTrue("Awaiting update", t1.isAlive()); assertTrue("Awaiting update", t2.isAlive()); while (t1.isAlive() || t2.isAlive()) { if (metadata.timeToNextUpdate(time) == 0) { metadata.update(TestUtils.singletonCluster(topic, 1), Collections.<String>emptySet(), time); time += refreshBackoffMs; } Thread.sleep(1); } t1.join(); t2.join(); assertFalse("No update needed.", metadata.timeToNextUpdate(time) == 0); time += metadataExpireMs; assertTrue("Update needed due to stale metadata.", metadata.timeToNextUpdate(time) == 0); }
public Metadata(long refreshBackoffMs, long metadataExpireMs, boolean allowAutoTopicCreation) { this(refreshBackoffMs, metadataExpireMs, allowAutoTopicCreation, false, new ClusterResourceListeners()); }
Metadata { public Metadata(long refreshBackoffMs, long metadataExpireMs, boolean allowAutoTopicCreation) { this(refreshBackoffMs, metadataExpireMs, allowAutoTopicCreation, false, new ClusterResourceListeners()); } }
Metadata { public Metadata(long refreshBackoffMs, long metadataExpireMs, boolean allowAutoTopicCreation) { this(refreshBackoffMs, metadataExpireMs, allowAutoTopicCreation, false, new ClusterResourceListeners()); } Metadata(long refreshBackoffMs, long metadataExpireMs, boolean allowAutoTopicCreation);  Metadata(long refreshBackoffMs, long metadataExpireMs, boolean allowAutoTopicCreation,
                    boolean topicExpiryEnabled, ClusterResourceListeners clusterResourceListeners); }
Metadata { public Metadata(long refreshBackoffMs, long metadataExpireMs, boolean allowAutoTopicCreation) { this(refreshBackoffMs, metadataExpireMs, allowAutoTopicCreation, false, new ClusterResourceListeners()); } 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 Metadata(long refreshBackoffMs, long metadataExpireMs, boolean allowAutoTopicCreation) { this(refreshBackoffMs, metadataExpireMs, allowAutoTopicCreation, false, new ClusterResourceListeners()); } 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; }