apache-kafka_0_9-lib/src/main/java/org/apache/kafka/common/security/kerberos/LoginManager.java
public static final LoginManager acquireLoginManager(LoginType loginType, Map<String, ?> configs) throws IOException, LoginException {
        synchronized (LoginManager.class) {
            LoginManager loginManager = CACHED_INSTANCES.get(loginType);
            if (loginManager == null) {
                loginManager = new LoginManager(loginType, configs);
                CACHED_INSTANCES.put(loginType, loginManager);
            }
            return loginManager.acquire();
        }
    }
public static final LoginManager acquireLoginManager(LoginType loginType, Map<String, ?> configs) throws IOException, LoginException {
        synchronized (LoginManager.class) {
            LoginManager loginManager = CACHED_INSTANCES.get(loginType);
            if (loginManager == null) {
                loginManager = new LoginManager(loginType, configs);
                CACHED_INSTANCES.put(loginType, loginManager);
            }
            return loginManager.acquire();
        }
    }
Returns an instance of `LoginManager` and increases its reference count.

`release()` should be invoked when the `LoginManager` is no longer needed. This method will try to reuse an
existing `LoginManager` for the provided `mode` if available. However, it expects `configs` to be the same for
every invocation and it will ignore them in the case where it's returning a cached instance of `LoginManager`.

This is a bit ugly and it would be nicer if we could pass the `LoginManager` to `ChannelBuilders.create` and
shut it down when the broker or clients are closed. It's straightforward to do the former, but it's more
complicated to do the latter without making the consumer API more complex.

@param loginType the type of the login context, it should be SERVER for the broker and CLIENT for the clients
(i.e. consumer and producer)
@param configs configuration as key/value pairs