/**
 * 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.pulsar.functions.utils.functioncache;

import java.io.File;
import java.io.IOException;
import java.net.URL;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.pulsar.functions.utils.Exceptions;

/**
 * An implementation of {@link FunctionCacheManager}.
 */
public class FunctionCacheManagerImpl implements FunctionCacheManager {

    /** Registered Functions **/
    private final Map<String, FunctionCacheEntry> cacheFunctions;

    public FunctionCacheManagerImpl() {
        this.cacheFunctions = Collections.synchronizedMap(new HashMap<>());
    }

    Map<String, FunctionCacheEntry> getCacheFunctions() {
        return cacheFunctions;
    }

    @Override
    public ClassLoader getClassLoader(String fid) {
        if (fid == null) {
            throw new NullPointerException("FunctionID not set");
        }

        synchronized (cacheFunctions) {
            FunctionCacheEntry entry = cacheFunctions.get(fid);
            if (entry == null) {
                throw new IllegalStateException("No dependencies are registered for function " + fid);
            }
            return entry.getClassLoader();
        }
    }

    @Override
    public void registerFunctionInstance(String fid,
                                         String eid,
                                         List<String> requiredJarFiles,
                                         List<URL> requiredClasspaths)
            throws IOException {
        if (fid == null) {
            throw new NullPointerException("FunctionID not set");
        }

        synchronized (cacheFunctions) {
            FunctionCacheEntry entry = cacheFunctions.get(fid);

            if (null == entry) {
                URL[] urls = new URL[requiredJarFiles.size() + requiredClasspaths.size()];
                int count = 0;
                try {
                    // add jar files to urls
                    for (String jarFile : requiredJarFiles) {
                        urls[count++] = new File(jarFile).toURI().toURL();
                    }

                    // add classpaths
                    for (URL url : requiredClasspaths) {
                        urls[count++] = url;
                    }

                    cacheFunctions.put(
                        fid,
                        new FunctionCacheEntry(
                            requiredJarFiles,
                            requiredClasspaths,
                            urls,
                            eid));
                } catch (Throwable cause) {
                    Exceptions.rethrowIOException(cause);
                }
            } else {
                entry.register(
                    eid,
                    requiredJarFiles,
                    requiredClasspaths);
            }
        }
    }

    @Override
    public void unregisterFunctionInstance(String fid,
                                           String eid) {
        synchronized (cacheFunctions) {
            FunctionCacheEntry entry = cacheFunctions.get(fid);

            if (null != entry) {
                if (entry.unregister(eid)) {
                    cacheFunctions.remove(fid);
                    entry.close();
                }
            }
        }
    }

    @Override
    public void close() {
        synchronized (cacheFunctions) {
            cacheFunctions.values().forEach(FunctionCacheEntry::close);
        }
    }


}

@@ -872,39 +872,42 @@
 ort 
-java.io.File;%0Aimport java.io.IO
+org.apache.pulsar.functions.utils.
 Exce
@@ -911,18 +911,20 @@
 xception
+s
 ;%0A
+%0A
 import j
@@ -931,15 +931,15 @@
 ava.
-net.URL
+io.File
 ;%0Aim
@@ -952,24 +952,22 @@
 ava.
-util.Collec
+io.IOExcep
 tion
-s
 ;%0Aim
@@ -980,20 +980,15 @@
 ava.
-util.HashMap
+net.URL
 ;%0Aim
@@ -1034,60 +1034,53 @@
 ap;%0A
-%0A
 import 
-org.apache.pulsar.functions.utils.Exceptions
+java.util.concurrent.ConcurrentHashMap
 ;%0A%0A/
@@ -1387,40 +1387,22 @@
 s = 
-Collections.synchronizedMap(new 
+new Concurrent
 Hash
@@ -1408,17 +1408,16 @@
 hMap%3C%3E()
-)
 ;%0A    %7D%0A

Switch to using ConcurrentHashMap (#1675)