/*
 * 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.flink.runtime.jobmaster;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.api.connector.source.Source;
import org.apache.flink.api.connector.source.SourceReader;
import org.apache.flink.api.connector.source.SourceReaderContext;
import org.apache.flink.api.connector.source.SourceSplit;
import org.apache.flink.api.connector.source.SplitEnumerator;
import org.apache.flink.api.connector.source.SplitEnumeratorContext;
import org.apache.flink.core.io.SimpleVersionedSerializer;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobGraphTestUtils;
import org.apache.flink.runtime.minicluster.MiniCluster;
import org.apache.flink.runtime.minicluster.MiniClusterConfiguration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
import org.apache.flink.util.TestLogger;

import org.junit.Test;

import static org.apache.flink.core.testutils.FlinkMatchers.containsMessage;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;

/** Integration tests for the {@link JobMaster}. */
public class JobMasterITCase extends TestLogger {

    @Test
    public void testRejectionOfEmptyJobGraphs() throws Exception {
        MiniCluster miniCluster =
                new MiniCluster(
                        new MiniClusterConfiguration.Builder()
                                .setNumTaskManagers(1)
                                .setNumSlotsPerTaskManager(1)
                                .build());
        miniCluster.start();
        JobGraph jobGraph = JobGraphTestUtils.emptyJobGraph();

        try {
            miniCluster.submitJob(jobGraph).get();
            fail("Expect failure");
        } catch (Throwable t) {
            assertThat(t, containsMessage("The given job is empty"));
        }
        miniCluster.close();
    }

    /**
     * This test is to guard against the issue reported in FLINK-22001, where any exception from the
     * JobManager initialization was not forwarded to the user.
     *
     * <p>TODO: This test relies on an internal error. Replace it with a more robust approach.
     */
    @Test
    public void testJobManagerInitializationExceptionsAreForwardedToTheUser() {
        // we must use the LocalStreamEnvironment to reproduce this issue.
        // It passes with the TestStreamEnvironment (which is initialized by the
        // MiniClusterResource). The LocalStreamEnvironment is polling the JobManager for the job
        // status, while TestStreamEnvironment is waiting on the resultFuture.
        StreamExecutionEnvironment see = StreamExecutionEnvironment.createLocalEnvironment();

        Source<String, MockSplit, Void> mySource = new FailOnInitializationSource();
        DataStream<String> stream =
                see.fromSource(mySource, WatermarkStrategy.noWatermarks(), "MySourceName");
        stream.addSink(new DiscardingSink<>());

        try {
            see.execute();
        } catch (Exception e) {
            assertThat(e, containsMessage("Context was not yet initialized"));
        }
    }

    private static class FailOnInitializationSource implements Source<String, MockSplit, Void> {
        @Override
        public Boundedness getBoundedness() {
            return Boundedness.CONTINUOUS_UNBOUNDED;
        }

        @Override
        public SourceReader<String, MockSplit> createReader(SourceReaderContext readerContext)
                throws Exception {
            throw new UnsupportedOperationException();
        }

        @Override
        public SplitEnumerator<MockSplit, Void> createEnumerator(
                SplitEnumeratorContext<MockSplit> enumContext) throws Exception {
            throw new RuntimeException();
        }

        @Override
        public SplitEnumerator<MockSplit, Void> restoreEnumerator(
                SplitEnumeratorContext<MockSplit> enumContext, Void checkpoint) throws Exception {
            throw new UnsupportedOperationException();
        }

        @Override
        public SimpleVersionedSerializer<MockSplit> getSplitSerializer() {
            throw new RuntimeException();
        }

        @Override
        public SimpleVersionedSerializer<Void> getEnumeratorCheckpointSerializer() {
            throw new UnsupportedOperationException();
        }
    }

    private static class MockSplit implements SourceSplit {
        @Override
        public String splitId() {
            throw new UnsupportedOperationException();
        }
    }
}

@@ -1528,35 +1528,33 @@
 ink.runtime.
-minicluster
+testutils
 .MiniCluster
@@ -1553,16 +1553,24 @@
 iCluster
+Resource
 ;%0Aimport
@@ -1599,19 +1599,17 @@
 ime.
-minicluster
+testutils
 .Min
@@ -1608,32 +1608,40 @@
 tils.MiniCluster
+Resource
 Configuration;%0Ai
@@ -2270,16 +2270,24 @@
 iCluster
+Resource
  miniClu
@@ -2324,16 +2324,24 @@
 iCluster
+Resource
 (%0A      
@@ -2373,16 +2373,24 @@
 iCluster
+Resource
 Configur
@@ -2444,16 +2444,19 @@
  .setNum
+ber
 TaskMana
@@ -2502,16 +2502,19 @@
  .setNum
+ber
 SlotsPer
@@ -2595,13 +2595,14 @@
 ter.
-start
+before
 ();%0A
@@ -2703,16 +2703,33 @@
 Cluster.
+getMiniCluster().
 submitJo
@@ -2890,25 +2890,39 @@
 );%0A        %7D
-%0A
+ finally %7B%0A    
         mini
@@ -2933,16 +2933,26 @@
 ter.
-close();
+after();%0A        %7D
 %0A   

use MiniClusterResource