[HUDI-7237] Hudi Streamer: Handle edge case with null schema, minor cleanups

[the-other-tim-brown]

Change Logs

• Reduces number of calls to schema provider by properly using orElseGet in one case and reusing existing writer config instead of recomputing
• Reuses existing hadoop configuration instance instead of creating new one when possible
• Handles case where schema provider returns null and there is an empty batch

Impact

• Reduces number of calls to schema providers which can reduce load on the provider or cost of accessing file system
• Fixes edge case where an empty batch could write a commit with a null value as the schema

Risk level (write none, low medium or high below)

low

Documentation Update

Describe any necessary documentation update if there is any new feature, config, or user-facing change

• The config description must be updated if new configs are added or the default value of the configs are changed
• Any new feature or user-facing change requires updating the Hudi website. Please create a Jira ticket, attach the
  ticket number here and follow the instruction to make
  changes to the website.

Contributor's checklist

• Read through contributor's guide
• Change Logs and Impact were stated clearly
• Adequate tests were added if applicable
• CI passed

[nsivabalan]

we call reInitWriteClient twice in this class. If we wer fixing the target schema here to go through getSchemaForWriteConfig, should we fix the other caller as well.
Also, looks like we are calling getSchemaForWriteConfig within reInitWriteClient(), should we remove that call if callers already take care of it.

[the-other-tim-brown]

We don't call getSchemaForWriteConfig from reInitWriteClient(). There is a call to getHoodieClientConfig that takes in a schema that must be non-null to hit the getSchemaForWriteConfig path though.

I think one way to simplify this is to make this call always happen if the schema is required in the config. When we initialize the StreamSync class, we will pass in some schema or null potentially but I don't think we really even need the writer schema for this case and want to avoid a read from the filesystem if possible to fetch the latest commit schema. If you agree, I can clean this up so that getHoodieClientConfig takes in a schema and a boolean indicating whether the schema must be set in the config.

[nsivabalan]

We do call.

reInitWriteClient {

       final HoodieWriteConfig initialWriteConfig = getHoodieClientConfig(targetSchema); 
}

within getHoodieClientConfig, we do call

    if (schema != null) {
      builder.withSchema(getSchemaForWriteConfig(schema).toString());
    }

but I agree, that if we had already called getSchemaForWriteConfig and fixed the target schema, it will be no-op. Was just pointing out we are making repeated calls.

[the-other-tim-brown]

@nsivabalan can you link the line where we make the call? I am missing something or have some outdated code somehow.

[nsivabalan]

hudi/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java

Line 1008 in 50f0d9f

private void reInitWriteClient(Schema sourceSchema, Schema targetSchema, Option<JavaRDD<HoodieRecord>> recordsOpt) throws IOException {

hudi/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java

Line 1014 in 50f0d9f

final HoodieWriteConfig initialWriteConfig = getHoodieClientConfig(targetSchema);

hudi/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java

Line 1049 in 50f0d9f

private HoodieWriteConfig getHoodieClientConfig(Schema schema) {

https://github.com/apache/hudi/blob/50f0d9f3baeafe92c38dfb0b59b337d83391ea42/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java#L1076C26-L1076C49

hudi/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java

Line 1111 in 50f0d9f

private Schema getSchemaForWriteConfig(Schema targetSchema) {

[the-other-tim-brown]

getHoodieClientConfig is also called from the constructor so you need to consider that path too if you are looking simply at paths that can eventually hit this getSchemaForWriteConfig method. I've updated the code so that there is no repeated call anymore and the call from the constructor avoids a potential schema lookup entirely.

[nsivabalan]

minor. do you think we can statically define once and use it rather than creating new objects everytime ?

[the-other-tim-brown]

yes, updating now

[the-other-tim-brown]

@hudi-bot run azure

[hudi-bot]

CI report:

• cb62ad9 UNKNOWN
• e83c0ad Azure: FAILURE

Bot commands

@hudi-bot supports the following commands:

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

[nsivabalan]