[HUDI-3528] Fix String convert issue and overwrite putAll method in Typed…

[stayrascal]

…Properties.java

Tips

• Thank you very much for contributing to Apache Hudi.
• Please review https://hudi.apache.org/contribute/how-to-contribute before opening a pull request.

What is the purpose of the pull request

I'm running Flink with Hudi in Java 11 environment(Even through I know Hudi might only support Java8/9/10), But we cannot change our JDK version. Based on this situation, We meet a NPE problem during HoodieWriteConfig.getIndexType(), but I test that it works in my local Java 8 environment. It seems that we implemented a TypedProperties class inherit from java.util.Properties class, and overwrite stringPropertyNames method & put method from this PR.

There are some difference of Properties.putAll(Map<?,?> t) between Java 8 and Java 11.

Java 8: TypedProperties will use HashTable's putAll method, which will call TypedProperties.put method, and the latter will add the incoming keys in TypedProperties.keys field.

public synchronized void putAll(Map<? extends K, ? extends V> t) {
        for (Map.Entry<? extends K, ? extends V> e : t.entrySet())
            put(e.getKey(), e.getValue());
    }

Java 11: Properties overwrite putAll method, and delegate the incoming map to ConcurrentHashMap. It won't call TypedProperties.put method anymore, which lead the TypedProperties.keys fields miss some keys, and then it cause that we miss the hoodie.index.type configuration.

@Override
    public synchronized void putAll(Map<?, ?> t) {
        map.putAll(t);
    }

So we also need to overwrite putAll method to better suitable high Java version runtime.

The exception trace with Java 11 runtime.

Caused by: java.lang.NullPointerException: Name is null
	at java.base/java.lang.Enum.valueOf(Enum.java:238)
	at org.apache.hudi.index.HoodieIndex$IndexType.valueOf(HoodieIndex.java:143)
	at org.apache.hudi.config.HoodieWriteConfig.getIndexType(HoodieWriteConfig.java:1347)
	at org.apache.hudi.index.FlinkHoodieIndexFactory.createIndex(FlinkHoodieIndexFactory.java:47)
	at org.apache.hudi.client.HoodieFlinkWriteClient.createIndex(HoodieFlinkWriteClient.java:105)
	at org.apache.hudi.client.BaseHoodieWriteClient.<init>(BaseHoodieWriteClient.java:145)
	at org.apache.hudi.client.BaseHoodieWriteClient.<init>(BaseHoodieWriteClient.java:131)
	at org.apache.hudi.client.HoodieFlinkWriteClient.<init>(HoodieFlinkWriteClient.java:96)
	at org.apache.hudi.util.StreamerUtil.createWriteClient(StreamerUtil.java:402)
	at org.apache.hudi.sink.StreamWriteOperatorCoordinator.start(StreamWriteOperatorCoordinator.java:166)
	at org.apache.flink.runtime.operators.coordination.OperatorCoordinatorHolder.start(OperatorCoordinatorHolder.java:194)
	at org.apache.flink.runtime.scheduler.DefaultOperatorCoordinatorHandler.startAllOperatorCoordinators(DefaultOperatorCoordinatorHandler.java:85)

Brief change log

• Overwrite putAll method in TypedPropertis
• Add class type checking before convert to String hardly

Verify this pull request

This change added tests and can be verified as follows:

• Add unit test case to test putAll cases in TestTypedProperties.java

Committer checklist

• Has a corresponding JIRA in PR title & commit

• Commit message is descriptive of the change

• CI is green

• Necessary doc changes done or have another open PR

• For large changes, please consider breaking it into sub-tasks under an umbrella JIRA.

[nsivabalan]

@stayrascal : this does not looks like minor. can you please file a jira and link it.

[nsivabalan]

CC @codope.

[hudi-bot]

CI report:

• 80b76f2 Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

• @hudi-bot run azure re-run the last Azure build

[stayrascal]

Thanks @nsivabalan review this PR. I failed to create a Jira card, not sure where I can create the card or I didn't have the permission. Instead, I create an issue here.

[nsivabalan]

LGTM

[codope]

lgtm.