[ENHANCEMENT] enhancement for setting up auth parameters

[wuzhanpeng]

Motivations

When I tried to create FlinkPulsarSource in the following way,

props.setProperty(PulsarOptions.AUTH_PLUGIN_CLASSNAME_KEY, "org.apache.pulsar.client.impl.auth.AuthenticationToken");
props.setProperty(PulsarOptions.AUTH_PARAMS_KEY, "token:abcdefghijklmn");

FlinkPulsarSource(adminUrl, clientConf, pulsarDeserializationSchema, props);

I found my flink job was failed and I can see some warnning logs in my broker like:

Failed to authenticate HTTP request: Authentication required

The reason for this problem is that only setting auth params in Properties will not take effect in the final initialization of PulsarClient.

Modifications

Add some extra checking for this issue.

[syhily]

This logic may not looks like right. We support both authParams and authMap. So you may need to check the authMap here.

[wuzhanpeng]

I add a check for authMap and PTAL~ @syhily

[syhily]

OK, LGTM. But this is only a hotfix we may perform a code cleanup and refactor these complex logic.