Skip to content

[HUDI-1200]: fixed NPE in CustomKeyGenerator #2093

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
vinothchandar merged 4 commits into from
Oct 21, 2020
+1 −2
Merged

[HUDI-1200]: fixed NPE in CustomKeyGenerator #2093

Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
99835a0
[HUDI-1200]: fixed NPE in CustomKeyGenerator
pratyakshsharma Sep 16, 2020
45e7ade
[HUDI-1200]: fixed build
pratyakshsharma Sep 16, 2020
a7232ef
[HUDI-1200]: retrigger build
pratyakshsharma Sep 17, 2020
0d63b55
[HUDI-1200]: fixed NPE in CustomKeyGenerator
pratyakshsharma Oct 4, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 10 additions & 9 deletions hudi-spark/src/main/java/org/apache/hudi/keygen/CustomKeyGenerator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@ public class CustomKeyGenerator extends BuiltinKeyGenerator {

private static final String DEFAULT_PARTITION_PATH_SEPARATOR = "/";
private static final String SPLIT_REGEX = ":";
private final TypedProperties props;

/**
* Used as a part of config in CustomKeyGenerator.java.
Expand All @@ -58,6 +59,7 @@ public enum PartitionKeyType {

public CustomKeyGenerator(TypedProperties props) {
super(props);
this.props = props;

@yanghua yanghua Sep 19, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hey, Since there is a protected config field in KeyGenerator, which can be shared with all subclasses. Why do not we check that must be non-nullable?

@vinothchandar vinothchandar Sep 22, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1 there is no need to duplicate this again in the sub class right.

@pratyakshsharma pratyakshsharma Sep 22, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@yanghua due to serialization and config field being transient, the issue is happening. In CustomKeyGenerator, we try to initialise new object of SimpleKeyGenerator and TimestampBasedKeyGenerator using config object, that is where NPE occurs. The only good way I can think of is to introduce a TimestampBasedKeyGenerator object in CustomKeyGenerator and call getPartitionPath method using this object. Let me do the changes here.

@bvaradar bvaradar Oct 2, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@pratyakshsharma : Why does config field need to be transient ?

@pratyakshsharma pratyakshsharma Oct 4, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@bvaradar It has been like this from the very beginning. I am not sure of the reason though. Making config field non-transient is the easiest fix possible here.

this.recordKeyFields = Arrays.stream(props.getString(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY()).split(",")).map(String::trim).collect(Collectors.toList());
this.partitionPathFields = Arrays.stream(props.getString(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY()).split(",")).map(String::trim).collect(Collectors.toList());
}
Expand Down Expand Up @@ -95,17 +97,17 @@ private String getPartitionPath(Option<GenericRecord> record, Option<Row> row) {
switch (keyType) {
case SIMPLE:
if (record.isPresent()) {
partitionPath.append(new SimpleKeyGenerator(config, partitionPathField).getPartitionPath(record.get()));
partitionPath.append(new SimpleKeyGenerator(props, partitionPathField).getPartitionPath(record.get()));
} else {
partitionPath.append(new SimpleKeyGenerator(config, partitionPathField).getPartitionPath(row.get()));
partitionPath.append(new SimpleKeyGenerator(props, partitionPathField).getPartitionPath(row.get()));
}
break;
case TIMESTAMP:
try {
if (record.isPresent()) {
partitionPath.append(new TimestampBasedKeyGenerator(config, partitionPathField).getPartitionPath(record.get()));
partitionPath.append(new TimestampBasedKeyGenerator(props, partitionPathField).getPartitionPath(record.get()));
} else {
partitionPath.append(new TimestampBasedKeyGenerator(config, partitionPathField).getPartitionPath(row.get()));
partitionPath.append(new TimestampBasedKeyGenerator(props, partitionPathField).getPartitionPath(row.get()));
}
} catch (IOException ioe) {
throw new HoodieDeltaStreamerException("Unable to initialise TimestampBasedKeyGenerator class");
Expand All @@ -118,24 +120,23 @@ private String getPartitionPath(Option<GenericRecord> record, Option<Row> row) {
partitionPath.append(DEFAULT_PARTITION_PATH_SEPARATOR);
}
partitionPath.deleteCharAt(partitionPath.length() - 1);

return partitionPath.toString();
}

@Override
public String getRecordKey(GenericRecord record) {
validateRecordKeyFields();
return getRecordKeyFields().size() == 1
? new SimpleKeyGenerator(config).getRecordKey(record)
: new ComplexKeyGenerator(config).getRecordKey(record);
? new SimpleKeyGenerator(props).getRecordKey(record)
: new ComplexKeyGenerator(props).getRecordKey(record);
}

@Override
public String getRecordKey(Row row) {
validateRecordKeyFields();
return getRecordKeyFields().size() == 1
? new SimpleKeyGenerator(config).getRecordKey(row)
: new ComplexKeyGenerator(config).getRecordKey(row);
? new SimpleKeyGenerator(props).getRecordKey(row)
: new ComplexKeyGenerator(props).getRecordKey(row);
}

private void validateRecordKeyFields() {
Expand Down