/**
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE
 * file distributed with this work for additional information regarding copyright ownership. The ASF licenses this file
 * to You under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the
 * License. You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
 * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations under the License.
 */
package org.apache.kafka.clients;

import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.test.TestUtils;
import org.junit.After;
import org.junit.Test;

import static org.junit.Assert.*;

public class MetadataTest {

    private long refreshBackoffMs = 100;
    private long metadataExpireMs = 1000;
    private Metadata metadata = new Metadata(refreshBackoffMs, metadataExpireMs);
    private AtomicBoolean backgroundError = new AtomicBoolean(false);
    
    @After
    public void tearDown() {
        assertFalse(backgroundError.get());
    }

    @Test
    public void testMetadata() throws Exception {
        long time = 0;
        metadata.update(Cluster.empty(), 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);
        Thread t2 = asyncFetch(topic);
        assertTrue("Awaiting update", t1.isAlive());
        assertTrue("Awaiting update", t2.isAlive());
        metadata.update(TestUtils.singletonCluster(topic, 1), time);
        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);
    }

    /**
     * Tests that {@link org.apache.kafka.clients.Metadata#awaitUpdate(int, long)} doesn't
     * wait forever with a max timeout value of 0
     *
     * @throws Exception
     * @see https://issues.apache.org/jira/browse/KAFKA-1836
     */
    @Test
    public void testMetadataUpdateWaitTime() throws Exception {
        long time = 0;
        metadata.update(Cluster.empty(), time);
        assertFalse("No update needed.", metadata.timeToNextUpdate(time) == 0);
        // first try with a max wait time of 0 and ensure that this returns back without waiting forever
        try {
            metadata.awaitUpdate(metadata.requestUpdate(), 0);
            fail("Wait on metadata update was expected to timeout, but it didn't");
        } catch (TimeoutException te) {
            // expected
        }
        // now try with a higher timeout value once
        final long twoSecondWait = 2000;
        try {
            metadata.awaitUpdate(metadata.requestUpdate(), twoSecondWait);
            fail("Wait on metadata update was expected to timeout, but it didn't");
        } catch (TimeoutException te) {
            // expected
        }
    }

    @Test
    public void testFailedUpdate() {
        long time = 100;
        metadata.update(Cluster.empty(), time);

        assertEquals(100, metadata.timeToNextUpdate(1000));
        metadata.failedUpdate(1100);

        assertEquals(100, metadata.timeToNextUpdate(1100));
        assertEquals(100, metadata.lastSuccessfulUpdate());

    }


    private Thread asyncFetch(final String topic) {
        Thread thread = new Thread() {
            public void run() {
                while (metadata.fetch().partitionsForTopic(topic) == null) {
                    try {
                        metadata.awaitUpdate(metadata.requestUpdate(), refreshBackoffMs);
                    } catch (Exception e) {
                        backgroundError.set(true);
                    }
                }
            }
        };
        thread.start();
        return thread;
    }
}

@@ -854,23 +854,25 @@
 c.Atomic
-Boolean
+Reference
 ;%0A%0Aimpor
@@ -1297,23 +1297,33 @@
 e Atomic
-Boolean
+Reference%3CString%3E
  backgro
@@ -1347,21 +1347,26 @@
 omic
-Boolean(false
+Reference%3CString%3E(
 );%0A 
@@ -1423,22 +1423,82 @@
   assert
-False(
+Null(%22Exception in background thread : %22 + backgroundError.get(), 
 backgrou
@@ -2236,68 +2236,435 @@
     
-metadata.update(TestUtils.singletonCluster(topic, 1), time);
+// Perform metadata update when an update is requested on the async fetch thread%0A        // This simulates the metadata update sequence in KafkaProducer%0A        while (t1.isAlive() %7C%7C t2.isAlive()) %7B%0A            if (metadata.timeToNextUpdate(time) == 0) %7B%0A                metadata.update(TestUtils.singletonCluster(topic, 1), time);%0A                time += refreshBackoffMs;%0A            %7D%0A            Thread.sleep(1);%0A        %7D
 %0A   
@@ -4835,12 +4835,20 @@
 set(
-true
+e.toString()
 );%0A 

Fix transient MetadataTest failure; reviewed by Jiangjie Qin and Guozhang Wang