You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by yi...@apache.org on 2024/02/27 07:35:54 UTC
(hudi) branch branch-0.x updated (23c9d85263b -> 855960f819e)
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a change to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
from 23c9d85263b [HUDI-7426] Fix logging issues in StreamSync (#10708)
new d361e80c083 [MINOR] Move release candidate validation to a separate GitHub action (#10729)
new ce0ee2f3fc4 [HUDI-7438] Add GitHub action to check Azure CI report (#10731)
new d0b34f08576 [HUDI-7275] Separate use of HoodieTimelineTimeZone.UTC and LOCAL in tests to prevent infinite loops (#10738)
new a64a0ed18e0 [HUDI-7440] Verify field exist in schema before fetching the value (#10733)
new b83f160e3f7 [HUDI-7433] Fix a bug in the HoodieBaseListData.isEmpty() empty-check logic (#10722) (#10722)
new 87e6e5e0991 [HUDI-7438] Add write permission of commit statuses in Azure CI check (#10745)
new cfbacf7b4cd [HUDI-7438] Reimplement Azure CI report check with open PRs (#10746)
new 41ee82827bb [HUDI-7438] Improve the filtering of PRs and pagination in scheduled workflow (#10747)
new d74c8cf432d [HUDI-7438] Fix workflow condition for issue_comment events (#10749)
new a61c5c015b6 [HUDI-7438] Fix issue number fetch in Azure CI check (#10751)
new 09d31136047 [MINOR] Fix typos in hudi-common module (#10748)
new 9b1f9952d19 [MINOR] StreamerUtil prints wrong table path (#10706)
new 7eb05e2d89d [HUDI-7416] Add interface for SourceProfile to be used in StreamSync for reading data (#10736)
new 3abccd14ef9 [MINOR] Update HoodieMetadataPayload bloom index error message (#10757)
new e5b28b68a65 [MINOR] Fix code style for HiveAvroSerializer (#10755)
new 413324346a7 [MINOR][DOCS] Update comment on hoodiemultitablestreamer (#10667)
new 5242b453236 [HUDI-7443] Fix decimal conversion with legacy bytes type (#10756)
new 855960f819e [HUDI-7441] Move `getWritePartitionPaths` method to common module to decouple hive dependency (#10744)
The 18 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.github/workflows/azure_ci.js | 85 ++++++++++++++++++
.github/workflows/azure_ci_check.yml | 80 +++++++++++++++++
.github/workflows/bot.yml | 72 ---------------
.github/workflows/release_candidate_validation.yml | 100 +++++++++++++++++++++
.github/workflows/scheduled_workflow.yml | 80 +++++++++++++++++
.../java/org/apache/hudi/avro/AvroSchemaUtils.java | 2 +-
.../java/org/apache/hudi/common/bloom/Key.java | 2 +-
.../hudi/common/data/HoodieBaseListData.java | 2 +-
.../util/queue/BaseHoodieQueueBasedExecutor.java | 2 +-
.../hudi/metadata/HoodieMetadataPayload.java | 2 +-
.../hudi/metadata/HoodieTableMetadataUtil.java | 15 +++-
.../hudi/common/data/TestHoodieListData.java | 22 +++++
.../apache/hudi/source/IncrementalInputSplits.java | 4 +-
.../java/org/apache/hudi/util/StreamerUtil.java | 2 +-
.../HoodieMergeOnReadTableInputFormat.java | 3 +-
.../hudi/hadoop/utils/HiveAvroSerializer.java | 28 +++---
.../hudi/hadoop/utils/HoodieInputFormatUtils.java | 13 ---
.../utils/HoodieRealtimeRecordReaderUtils.java | 26 +++---
.../hudi/MergeOnReadIncrementalRelation.scala | 3 +-
.../org/apache/hudi/TestHoodieSparkSqlWriter.scala | 59 ++----------
.../apache/hudi/TestHoodieSparkSqlWriterUtc.scala | 85 ++++++++++++++++++
.../org/apache/hudi/utilities/UtilHelpers.java | 19 ++++
.../hudi/utilities/deltastreamer/DeltaSync.java | 4 +-
.../hudi/utilities/sources/AvroKafkaSource.java | 13 ++-
.../hudi/utilities/sources/JsonKafkaSource.java | 14 ++-
.../apache/hudi/utilities/sources/KafkaSource.java | 39 +++++---
.../hudi/utilities/sources/ProtoKafkaSource.java | 15 +++-
.../org/apache/hudi/utilities/sources/Source.java | 11 ++-
.../utilities/sources/helpers/KafkaOffsetGen.java | 39 ++++----
.../utilities/streamer/DefaultStreamContext.java} | 31 ++++---
.../streamer/HoodieMultiTableStreamer.java | 2 +-
.../hudi/utilities/streamer/HoodieStreamer.java | 21 +++--
.../hudi/utilities/streamer/SourceProfile.java | 34 ++++---
.../utilities/streamer/SourceProfileSupplier.java | 18 ++--
.../hudi/utilities/streamer/StreamContext.java | 28 +++---
.../apache/hudi/utilities/streamer/StreamSync.java | 10 +--
.../utilities/sources/BaseTestKafkaSource.java | 51 +++++++++++
.../utilities/sources/TestJsonKafkaSource.java | 15 +++-
.../utilities/sources/TestProtoKafkaSource.java | 3 +-
39 files changed, 768 insertions(+), 286 deletions(-)
create mode 100644 .github/workflows/azure_ci.js
create mode 100644 .github/workflows/azure_ci_check.yml
create mode 100644 .github/workflows/release_candidate_validation.yml
create mode 100644 .github/workflows/scheduled_workflow.yml
create mode 100644 hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriterUtc.scala
copy hudi-utilities/src/{test/java/org/apache/hudi/utilities/sources/helpers/SchemaTestProvider.java => main/java/org/apache/hudi/utilities/streamer/DefaultStreamContext.java} (53%)
copy hudi-io/src/main/java/org/apache/hudi/storage/StoragePathFilter.java => hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/SourceProfile.java (50%)
copy hudi-io/src/main/java/org/apache/hudi/storage/StoragePathFilter.java => hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/SourceProfileSupplier.java (68%)
copy hudi-common/src/main/java/org/apache/hudi/common/conflict/detection/EarlyConflictDetectionStrategy.java => hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamContext.java (55%)
(hudi) 06/18: [HUDI-7438] Add write permission of commit statuses in Azure CI check (#10745)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 87e6e5e0991ec051ac8219b586f62aedd30232a4
Author: Y Ethan Guo <et...@gmail.com>
AuthorDate: Sat Feb 24 10:55:32 2024 -0800
[HUDI-7438] Add write permission of commit statuses in Azure CI check (#10745)
---
.github/workflows/azure_ci_check.yml | 1 +
1 file changed, 1 insertion(+)
diff --git a/.github/workflows/azure_ci_check.yml b/.github/workflows/azure_ci_check.yml
index 347d9c2959f..3c4ba58a7f3 100644
--- a/.github/workflows/azure_ci_check.yml
+++ b/.github/workflows/azure_ci_check.yml
@@ -22,6 +22,7 @@ on:
types: [ created, edited, deleted ]
permissions:
+ statuses: write
pull-requests: read
issues: read
(hudi) 08/18: [HUDI-7438] Improve the filtering of PRs and pagination in scheduled workflow (#10747)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 41ee82827bb777d0ed0e8fd4a4a77be400b63b5d
Author: Y Ethan Guo <et...@gmail.com>
AuthorDate: Sat Feb 24 21:53:39 2024 -0800
[HUDI-7438] Improve the filtering of PRs and pagination in scheduled workflow (#10747)
---
.github/workflows/azure_ci.js | 11 +++++++----
.github/workflows/scheduled_workflow.yml | 16 ++++++++++------
2 files changed, 17 insertions(+), 10 deletions(-)
diff --git a/.github/workflows/azure_ci.js b/.github/workflows/azure_ci.js
index 98ba39488b0..737b8db9917 100644
--- a/.github/workflows/azure_ci.js
+++ b/.github/workflows/azure_ci.js
@@ -21,21 +21,24 @@ async function checkAzureCiAndCreateCommitStatus({ github, context, prNumber, la
console.log(`- Checking Azure CI status of PR: ${prNumber} ${latestCommitHash}`);
const botUsername = 'hudi-bot';
- const comments = await github.rest.issues.listComments({
+ const comments = await github.paginate(github.rest.issues.listComments, {
owner: context.repo.owner,
repo: context.repo.repo,
issue_number: prNumber,
+ sort: 'updated',
+ direction: 'desc',
+ per_page: 100
});
// Find the latest comment from hudi-bot containing the Azure CI report
- const botComments = comments.data.filter(comment => comment.user.login === botUsername);
- const lastComment = botComments.pop();
+ const botComments = comments.filter(comment => comment.user.login === botUsername);
let status = 'pending';
let message = 'In progress';
let azureRunLink = '';
- if (lastComment) {
+ if (botComments.length > 0) {
+ const lastComment = botComments[0];
const reportPrefix = `${latestCommitHash} Azure: `
const successReportString = `${reportPrefix}[SUCCESS]`
const failureReportString = `${reportPrefix}[FAILURE]`
diff --git a/.github/workflows/scheduled_workflow.yml b/.github/workflows/scheduled_workflow.yml
index 39d291fed40..4e17ee12990 100644
--- a/.github/workflows/scheduled_workflow.yml
+++ b/.github/workflows/scheduled_workflow.yml
@@ -44,16 +44,20 @@ jobs:
with:
github-token: ${{secrets.GITHUB_TOKEN}}
script: |
- const since = new Date(new Date().getTime() - (330 * 1000)).toISOString();
- const query = `repo:${context.repo.owner}/${context.repo.repo} type:pr updated:>=${since}`;
- const response = await github.rest.search.issuesAndPullRequests({
- q: query
+ // Cron schedule may not be reliable so giving buffer time to avoid missing recent PRs
+ const since = new Date(new Date().getTime() - (900 * 1000)).toISOString();
+ const query = `repo:${context.repo.owner}/${context.repo.repo} type:pr state:open base:master updated:>=${since}`;
+ const openPrs = await github.paginate(github.rest.search.issuesAndPullRequests, {
+ q: query,
+ sort: 'updated',
+ order: 'desc',
+ per_page: 100
});
- // Filter for open PRs
- const openPrs = response.data.items.filter(pr => pr.state === 'open');
const checkAzureCiAndCreateCommitStatus = require(`${process.env.GITHUB_WORKSPACE}/.github/workflows/azure_ci.js`);
+ console.log(`Number of PRs to process: ${openPrs.length}`);
+
for (const pr of openPrs) {
console.log(`*** Processing PR: ${pr.title}, URL: ${pr.html_url}`);
(hudi) 14/18: [MINOR] Update HoodieMetadataPayload bloom index error message (#10757)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 3abccd14ef942e64d62157393ae49f4e67f28165
Author: Sagar Sumit <sa...@gmail.com>
AuthorDate: Tue Feb 27 03:53:16 2024 +0530
[MINOR] Update HoodieMetadataPayload bloom index error message (#10757)
---
.../src/main/java/org/apache/hudi/metadata/HoodieMetadataPayload.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataPayload.java b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataPayload.java
index e0fd3dd4bfd..483e00ba734 100644
--- a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataPayload.java
+++ b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataPayload.java
@@ -217,7 +217,7 @@ public class HoodieMetadataPayload implements HoodieRecordPayload<HoodieMetadata
// Otherwise, it has to be present or the record would be considered invalid
if (bloomFilterRecord == null) {
checkArgument(record.getSchema().getField(SCHEMA_FIELD_ID_BLOOM_FILTER) == null,
- String.format("Valid %s record expected for type: %s", SCHEMA_FIELD_ID_BLOOM_FILTER, METADATA_TYPE_COLUMN_STATS));
+ String.format("Valid %s record expected for type: %s", SCHEMA_FIELD_ID_BLOOM_FILTER, METADATA_TYPE_BLOOM_FILTER));
} else {
bloomFilterMetadata = new HoodieMetadataBloomFilter(
(String) bloomFilterRecord.get(BLOOM_FILTER_FIELD_TYPE),
(hudi) 12/18: [MINOR] StreamerUtil prints wrong table path (#10706)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 9b1f9952d19a8baf0d6a202546a369629747cfed
Author: zhuanshenbsj1 <34...@users.noreply.github.com>
AuthorDate: Mon Feb 26 03:09:50 2024 +0800
[MINOR] StreamerUtil prints wrong table path (#10706)
* update print
* Update hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java
Co-authored-by: Y Ethan Guo <et...@gmail.com>
---------
Co-authored-by: Y Ethan Guo <et...@gmail.com>
---
.../hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java
index 648a108d867..40519ae4ed7 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java
@@ -229,7 +229,7 @@ public class StreamerUtil {
LOG.info("Table initialized under base path {}", basePath);
return metaClient;
} else {
- LOG.info("Table [{}/{}] already exists, no need to initialize the table",
+ LOG.info("Table [path={}, name={}] already exists, no need to initialize the table",
basePath, conf.getString(FlinkOptions.TABLE_NAME));
return StreamerUtil.createMetaClient(basePath, hadoopConf);
}
(hudi) 05/18: [HUDI-7433] Fix a bug in the HoodieBaseListData.isEmpty() empty-check logic (#10722) (#10722)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit b83f160e3f75a27aef6d91d7f093e7be070eb4e4
Author: usberkeley <15...@users.noreply.github.com>
AuthorDate: Sat Feb 24 12:42:46 2024 +0800
[HUDI-7433] Fix a bug in the HoodieBaseListData.isEmpty() empty-check logic (#10722) (#10722)
---
.../hudi/common/data/HoodieBaseListData.java | 2 +-
.../hudi/common/data/TestHoodieListData.java | 22 ++++++++++++++++++++++
2 files changed, 23 insertions(+), 1 deletion(-)
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/data/HoodieBaseListData.java b/hudi-common/src/main/java/org/apache/hudi/common/data/HoodieBaseListData.java
index 7bc276b36e6..6f3dbfcef99 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/data/HoodieBaseListData.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/data/HoodieBaseListData.java
@@ -53,7 +53,7 @@ public abstract class HoodieBaseListData<T> {
protected boolean isEmpty() {
if (lazy) {
- return data.asLeft().findAny().isPresent();
+ return !data.asLeft().findAny().isPresent();
} else {
return data.asRight().isEmpty();
}
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/data/TestHoodieListData.java b/hudi-common/src/test/java/org/apache/hudi/common/data/TestHoodieListData.java
index ea19f128d1a..795318f5e01 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/data/TestHoodieListData.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/data/TestHoodieListData.java
@@ -27,12 +27,15 @@ import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
class TestHoodieListData {
@@ -72,4 +75,23 @@ class TestHoodieListData {
IntStream.rangeClosed(0, 100).boxed().collect(Collectors.toList()));
assertEquals(1, listData.getNumPartitions());
}
+
+ @Test
+ public void testIsEmpty() {
+ // HoodieListData bearing eager execution semantic
+ HoodieData<Integer> listData = HoodieListData.eager(
+ IntStream.rangeClosed(0, 100).boxed().collect(Collectors.toList()));
+ assertFalse(listData.isEmpty());
+
+ HoodieData<Integer> emptyListData = HoodieListData.eager(Collections.emptyList());
+ assertTrue(emptyListData.isEmpty());
+
+ // HoodieListData bearing lazy execution semantic
+ listData = HoodieListData.lazy(
+ IntStream.rangeClosed(0, 100).boxed().collect(Collectors.toList()));
+ assertFalse(listData.isEmpty());
+
+ emptyListData = HoodieListData.lazy(Collections.emptyList());
+ assertTrue(emptyListData.isEmpty());
+ }
}
(hudi) 04/18: [HUDI-7440] Verify field exist in schema before fetching the value (#10733)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit a64a0ed18e0583703378ae5efd5edf0048dc6588
Author: Mani Chandrasekar <ch...@amazon.com>
AuthorDate: Fri Feb 23 20:39:16 2024 -0800
[HUDI-7440] Verify field exist in schema before fetching the value (#10733)
---
.../hudi/hadoop/utils/HoodieRealtimeRecordReaderUtils.java | 10 +++++-----
1 file changed, 5 insertions(+), 5 deletions(-)
diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeRecordReaderUtils.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeRecordReaderUtils.java
index 539bc21eb88..35fa7966c59 100644
--- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeRecordReaderUtils.java
+++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeRecordReaderUtils.java
@@ -26,7 +26,6 @@ import org.apache.hudi.hadoop.config.HoodieRealtimeConfig;
import org.apache.hudi.io.storage.HoodieFileReader;
import org.apache.hudi.io.storage.HoodieFileReaderFactory;
-import org.apache.avro.AvroRuntimeException;
import org.apache.avro.JsonProperties;
import org.apache.avro.LogicalType;
import org.apache.avro.LogicalTypes;
@@ -195,12 +194,13 @@ public class HoodieRealtimeRecordReaderUtils {
Writable[] recordValues = new Writable[schema.getFields().size()];
int recordValueIndex = 0;
for (Schema.Field field : schema.getFields()) {
- // TODO Revisit Avro exception handling in future
Object fieldValue = null;
- try {
+ if (record.getSchema().getField(field.name()) != null) {
fieldValue = record.get(field.name());
- } catch (AvroRuntimeException e) {
- LOG.debug("Field:" + field.name() + "not found in Schema:" + schema);
+ } else {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Field:" + field.name() + "not found in Schema:" + schema);
+ }
}
recordValues[recordValueIndex++] = avroToArrayWritable(fieldValue, field.schema(), supportTimestamp);
}
(hudi) 07/18: [HUDI-7438] Reimplement Azure CI report check with open PRs (#10746)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit cfbacf7b4cdd7c20493ce69ecb27c19620a6fdca
Author: Y Ethan Guo <et...@gmail.com>
AuthorDate: Sat Feb 24 20:00:27 2024 -0800
[HUDI-7438] Reimplement Azure CI report check with open PRs (#10746)
---
.github/workflows/azure_ci.js | 82 +++++++++++++++++++++++++++++
.github/workflows/azure_ci_check.yml | 90 ++++++++++++++------------------
.github/workflows/scheduled_workflow.yml | 76 +++++++++++++++++++++++++++
3 files changed, 197 insertions(+), 51 deletions(-)
diff --git a/.github/workflows/azure_ci.js b/.github/workflows/azure_ci.js
new file mode 100644
index 00000000000..98ba39488b0
--- /dev/null
+++ b/.github/workflows/azure_ci.js
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+async function checkAzureCiAndCreateCommitStatus({ github, context, prNumber, latestCommitHash }) {
+ console.log(`- Checking Azure CI status of PR: ${prNumber} ${latestCommitHash}`);
+ const botUsername = 'hudi-bot';
+
+ const comments = await github.rest.issues.listComments({
+ owner: context.repo.owner,
+ repo: context.repo.repo,
+ issue_number: prNumber,
+ });
+
+ // Find the latest comment from hudi-bot containing the Azure CI report
+ const botComments = comments.data.filter(comment => comment.user.login === botUsername);
+ const lastComment = botComments.pop();
+
+ let status = 'pending';
+ let message = 'In progress';
+ let azureRunLink = '';
+
+ if (lastComment) {
+ const reportPrefix = `${latestCommitHash} Azure: `
+ const successReportString = `${reportPrefix}[SUCCESS]`
+ const failureReportString = `${reportPrefix}[FAILURE]`
+
+ if (lastComment.body.includes(reportPrefix)) {
+ if (lastComment.body.includes(successReportString)) {
+ message = 'Successful on the latest commit';
+ status = 'success';
+ } else if (lastComment.body.includes(failureReportString)) {
+ message = 'Failed on the latest commit';
+ status = 'failure';
+ }
+ }
+
+ const linkRegex = /\[[a-zA-Z]+\]\((https?:\/\/[^\s]+)\)/;
+ const parts = lastComment.body.split(reportPrefix);
+ const secondPart = parts.length > 1 ? parts[1] : '';
+ const match = secondPart.match(linkRegex);
+
+ if (match) {
+ azureRunLink = match[1];
+ }
+ }
+
+ console.log(`Status: ${status}`);
+ console.log(`Azure Run Link: ${azureRunLink}`);
+ console.log(`${message}`);
+
+ console.log(`- Create commit status of PR based on Azure CI status: ${prNumber} ${latestCommitHash}`);
+ // Create or update the commit status for Azure CI
+ await github.rest.repos.createCommitStatus({
+ owner: context.repo.owner,
+ repo: context.repo.repo,
+ sha: latestCommitHash,
+ state: status,
+ target_url: azureRunLink,
+ description: message,
+ context: 'Azure CI'
+ });
+
+ return { status, message, azureRunLink };
+}
+
+module.exports = checkAzureCiAndCreateCommitStatus;
diff --git a/.github/workflows/azure_ci_check.yml b/.github/workflows/azure_ci_check.yml
index 3c4ba58a7f3..17484a40aa5 100644
--- a/.github/workflows/azure_ci_check.yml
+++ b/.github/workflows/azure_ci_check.yml
@@ -27,67 +27,55 @@ permissions:
issues: read
jobs:
- check-azure-ci-report:
- if: "!contains(github.event.pull_request.body, 'HOTFIX: SKIP AZURE CI')"
+ check-azure-ci-and-add-commit-status:
+ if: |
+ github.event.issue.pull_request != null &&
+ github.event.issue.pull_request != '' &&
+ github.event.issue_comment.user.login == 'hudi-bot'
runs-on: ubuntu-latest
steps:
- - name: Get last commit hash
- id: last_commit
- uses: actions/github-script@v7
- with:
- github-token: ${{secrets.GITHUB_TOKEN}}
- script: |
- const pr = context.payload.pull_request;
- const lastCommitHash = pr.head.sha;
- console.log(`Last commit hash: ${lastCommitHash}`);
- // Set the output variable to be used in subsequent step
- core.setOutput("last_commit_hash", lastCommitHash);
+ - name: Checkout repository
+ uses: actions/checkout@v2
- - name: Check Azure CI report in PR comment
+ - name: Check PR state
+ id: check_pr_state
uses: actions/github-script@v7
with:
github-token: ${{secrets.GITHUB_TOKEN}}
script: |
- const lastCommitHash = '${{ steps.last_commit.outputs.last_commit_hash }}'
- const botUsername = 'hudi-bot';
-
- const issueNumber = context.payload.pull_request.number;
- const comments = await github.rest.issues.listComments({
+ const issueNumber = github.event.issue.number;
+ const { data: pullRequest } = await github.rest.pulls.get({
owner: context.repo.owner,
repo: context.repo.repo,
- issue_number: issueNumber,
+ pull_number: issueNumber
});
- // Find the last comment from hudi-bot containing the Azure CI report
- const botComments = comments.data.filter(comment => comment.user.login === botUsername);
- const lastComment = botComments.pop();
+ // Only check open PRs and a PR that is not a HOTFIX
+ const shouldSkip = (pullRequest.body.includes('HOTFIX: SKIP AZURE CI')
+ || pullRequest.state != 'open');
- if (lastComment) {
- const reportPrefix = '${lastCommitHash} Azure: '
- const successReportString = '${reportPrefix}[SUCCESS]'
- const failureReportString = '${reportPrefix}[FAILURE]'
- if (lastComment.body.includes(reportPrefix)) {
- if (lastComment.body.includes(successReportString)) {
- console.log(`Azure CI succeeded on the latest commit of the PR.`);
- return true;
- } else if (lastComment.body.includes(failureReportString)) {
- console.log(`Azure CI failed on the latest commit of the PR.`);
- core.setFailed("Azure CI failed on the latest commit of the PR.");
- return false;
- } else {
- console.log(`Azure CI is in progress on the latest commit of the PR.`);
- core.setFailed("Azure CI is in progress on the latest commit of the PR.");
- return false;
- }
- } else {
- console.log(`No Azure CI report on the latest commit of the PR.`);
- core.setFailed("No Azure CI report on the latest commit of the PR.");
- return false;
- }
- } else {
- console.log(`Azure CI report does not seem to be ready yet.`);
- core.setFailed("Azure CI report does not seem to be ready yet.");
- return false;
+ if (!shouldSkip) {
+ const commitHash = pullRequest.head.sha;
+ console.log(`Latest commit hash: ${commitHash}`);
+ // Set the output variable to be used in subsequent step
+ core.setOutput("latest_commit_hash", commitHash);
}
- env:
- GITHUB_TOKEN: ${{secrets.GITHUB_TOKEN}}
+ console.log(`Should skip Azure CI? ${shouldSkip}`);
+ return shouldSkip;
+
+ - name: Check Azure CI report and create commit status to PR
+ if: steps.check_pr_state.outputs.result != 'true'
+ uses: actions/github-script@v7
+ with:
+ github-token: ${{secrets.GITHUB_TOKEN}}
+ script: |
+ const latestCommitHash = '${{ steps.check_pr_state.outputs.latest_commit_hash }}'
+ const issueNumber = github.event.issue.number;
+ const checkAzureCiAndCreateCommitStatus = require(`${process.env.GITHUB_WORKSPACE}/.github/workflows/azure_ci.js`);
+
+ await checkAzureCiAndCreateCommitStatus({
+ github,
+ context,
+ prNumber: issueNumber,
+ latestCommitHash: latestCommitHash
+ });
diff --git a/.github/workflows/scheduled_workflow.yml b/.github/workflows/scheduled_workflow.yml
new file mode 100644
index 00000000000..39d291fed40
--- /dev/null
+++ b/.github/workflows/scheduled_workflow.yml
@@ -0,0 +1,76 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+name: Scheduled Workflow
+
+on:
+ schedule:
+ # Runs every 5 minutes
+ - cron: '*/5 * * * *'
+
+permissions:
+ statuses: write
+ pull-requests: read
+ issues: read
+
+jobs:
+ process-new-and-updated-prs:
+ runs-on: ubuntu-latest
+ steps:
+ - name: Checkout repository
+ uses: actions/checkout@v2
+
+ - name: Process new and updated PRs
+ # We have to run any actions that require write permissions here
+ # since the workflow triggered by events from a PR in a fork
+ # (not apache/hudi but other_owner/hudi) does not run on a
+ # GITHUB_TOKEN with write permissions (this is prohibited by
+ # Apache).
+ uses: actions/github-script@v7
+ with:
+ github-token: ${{secrets.GITHUB_TOKEN}}
+ script: |
+ const since = new Date(new Date().getTime() - (330 * 1000)).toISOString();
+ const query = `repo:${context.repo.owner}/${context.repo.repo} type:pr updated:>=${since}`;
+ const response = await github.rest.search.issuesAndPullRequests({
+ q: query
+ });
+
+ // Filter for open PRs
+ const openPrs = response.data.items.filter(pr => pr.state === 'open');
+ const checkAzureCiAndCreateCommitStatus = require(`${process.env.GITHUB_WORKSPACE}/.github/workflows/azure_ci.js`);
+
+ for (const pr of openPrs) {
+ console.log(`*** Processing PR: ${pr.title}, URL: ${pr.html_url}`);
+
+ if (!pr.body.includes('HOTFIX: SKIP AZURE CI')) {
+ const { data: pullRequest } = await github.rest.pulls.get({
+ owner: context.repo.owner,
+ repo: context.repo.repo,
+ pull_number: pr.number
+ });
+ const latestCommitHash = pullRequest.head.sha;
+
+ // Create commit status based on Azure CI report to PR
+ await checkAzureCiAndCreateCommitStatus({
+ github,
+ context,
+ prNumber: pr.number,
+ latestCommitHash: latestCommitHash
+ });
+ }
+ }
(hudi) 15/18: [MINOR] Fix code style for HiveAvroSerializer (#10755)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit e5b28b68a65933d2fdf097de9931d1285ba76724
Author: stream2000 <st...@apache.org>
AuthorDate: Tue Feb 27 06:47:20 2024 +0800
[MINOR] Fix code style for HiveAvroSerializer (#10755)
---
.../hudi/hadoop/utils/HiveAvroSerializer.java | 22 +++++++++++-----------
1 file changed, 11 insertions(+), 11 deletions(-)
diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HiveAvroSerializer.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HiveAvroSerializer.java
index a0d1b086e03..5f33844d60c 100644
--- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HiveAvroSerializer.java
+++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HiveAvroSerializer.java
@@ -99,7 +99,7 @@ public class HiveAvroSerializer {
List<? extends StructField> allStructFieldRefs = soi.getAllStructFieldRefs();
List<Object> structFieldsDataAsList = soi.getStructFieldsDataAsList(o);
- for (int i = 0; i < size; i++) {
+ for (int i = 0; i < size; i++) {
Schema.Field field = schema.getFields().get(i);
if (i >= columnTypes.size()) {
break;
@@ -134,7 +134,7 @@ public class HiveAvroSerializer {
* Determine if an Avro schema is of type Union[T, NULL]. Avro supports nullable
* types via a union of type T and null. This is a very common use case.
* As such, we want to silently convert it to just T and allow the value to be null.
- *
+ * <p>
* When a Hive union type is used with AVRO, the schema type becomes
* Union[NULL, T1, T2, ...]. The NULL in the union should be silently removed
*
@@ -266,7 +266,7 @@ public class HiveAvroSerializer {
GenericData.Record record = new GenericData.Record(schema);
ArrayList<TypeInfo> allStructFieldTypeInfos = typeInfo.getAllStructFieldTypeInfos();
- for (int i = 0; i < size; i++) {
+ for (int i = 0; i < size; i++) {
Schema.Field field = schema.getFields().get(i);
setUpRecordFieldFromWritable(allStructFieldTypeInfos.get(i), structFieldsDataAsList.get(i),
allStructFieldRefs.get(i).getFieldObjectInspector(), record, field);
@@ -278,26 +278,26 @@ public class HiveAvroSerializer {
switch (fieldOI.getPrimitiveCategory()) {
case BINARY:
if (schema.getType() == Schema.Type.BYTES) {
- return AvroSerdeUtils.getBufferFromBytes((byte[])fieldOI.getPrimitiveJavaObject(structFieldData));
+ return AvroSerdeUtils.getBufferFromBytes((byte[]) fieldOI.getPrimitiveJavaObject(structFieldData));
} else if (schema.getType() == Schema.Type.FIXED) {
- GenericData.Fixed fixed = new GenericData.Fixed(schema, (byte[])fieldOI.getPrimitiveJavaObject(structFieldData));
+ GenericData.Fixed fixed = new GenericData.Fixed(schema, (byte[]) fieldOI.getPrimitiveJavaObject(structFieldData));
return fixed;
} else {
throw new HoodieException("Unexpected Avro schema for Binary TypeInfo: " + schema.getType());
}
case DECIMAL:
- HiveDecimal dec = (HiveDecimal)fieldOI.getPrimitiveJavaObject(structFieldData);
- LogicalTypes.Decimal decimal = (LogicalTypes.Decimal)schema.getLogicalType();
+ HiveDecimal dec = (HiveDecimal) fieldOI.getPrimitiveJavaObject(structFieldData);
+ LogicalTypes.Decimal decimal = (LogicalTypes.Decimal) schema.getLogicalType();
BigDecimal bd = new BigDecimal(dec.toString()).setScale(decimal.getScale());
return HoodieAvroUtils.DECIMAL_CONVERSION.toFixed(bd, schema, decimal);
case CHAR:
- HiveChar ch = (HiveChar)fieldOI.getPrimitiveJavaObject(structFieldData);
+ HiveChar ch = (HiveChar) fieldOI.getPrimitiveJavaObject(structFieldData);
return new Utf8(ch.getStrippedValue());
case VARCHAR:
- HiveVarchar vc = (HiveVarchar)fieldOI.getPrimitiveJavaObject(structFieldData);
+ HiveVarchar vc = (HiveVarchar) fieldOI.getPrimitiveJavaObject(structFieldData);
return new Utf8(vc.getValue());
case STRING:
- String string = (String)fieldOI.getPrimitiveJavaObject(structFieldData);
+ String string = (String) fieldOI.getPrimitiveJavaObject(structFieldData);
return new Utf8(string);
case DATE:
return HoodieHiveUtils.getDays(structFieldData);
@@ -364,7 +364,7 @@ public class HiveAvroSerializer {
ObjectInspector mapValueObjectInspector = fieldOI.getMapValueObjectInspector();
TypeInfo mapKeyTypeInfo = typeInfo.getMapKeyTypeInfo();
TypeInfo mapValueTypeInfo = typeInfo.getMapValueTypeInfo();
- Map<?,?> map = fieldOI.getMap(structFieldData);
+ Map<?, ?> map = fieldOI.getMap(structFieldData);
Schema valueType = schema.getValueType();
Map<Object, Object> deserialized = new LinkedHashMap<Object, Object>(fieldOI.getMapSize(structFieldData));
(hudi) 02/18: [HUDI-7438] Add GitHub action to check Azure CI report (#10731)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit ce0ee2f3fc4c8ced6cef4afb7ba6966b5657d225
Author: Y Ethan Guo <et...@gmail.com>
AuthorDate: Thu Feb 22 21:44:14 2024 -0800
[HUDI-7438] Add GitHub action to check Azure CI report (#10731)
---
.github/workflows/azure_ci_check.yml | 92 ++++++++++++++++++++++++++++++++++++
1 file changed, 92 insertions(+)
diff --git a/.github/workflows/azure_ci_check.yml b/.github/workflows/azure_ci_check.yml
new file mode 100644
index 00000000000..347d9c2959f
--- /dev/null
+++ b/.github/workflows/azure_ci_check.yml
@@ -0,0 +1,92 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+name: Azure CI
+
+on:
+ issue_comment:
+ types: [ created, edited, deleted ]
+
+permissions:
+ pull-requests: read
+ issues: read
+
+jobs:
+ check-azure-ci-report:
+ if: "!contains(github.event.pull_request.body, 'HOTFIX: SKIP AZURE CI')"
+ runs-on: ubuntu-latest
+ steps:
+ - name: Get last commit hash
+ id: last_commit
+ uses: actions/github-script@v7
+ with:
+ github-token: ${{secrets.GITHUB_TOKEN}}
+ script: |
+ const pr = context.payload.pull_request;
+ const lastCommitHash = pr.head.sha;
+ console.log(`Last commit hash: ${lastCommitHash}`);
+ // Set the output variable to be used in subsequent step
+ core.setOutput("last_commit_hash", lastCommitHash);
+
+ - name: Check Azure CI report in PR comment
+ uses: actions/github-script@v7
+ with:
+ github-token: ${{secrets.GITHUB_TOKEN}}
+ script: |
+ const lastCommitHash = '${{ steps.last_commit.outputs.last_commit_hash }}'
+ const botUsername = 'hudi-bot';
+
+ const issueNumber = context.payload.pull_request.number;
+ const comments = await github.rest.issues.listComments({
+ owner: context.repo.owner,
+ repo: context.repo.repo,
+ issue_number: issueNumber,
+ });
+
+ // Find the last comment from hudi-bot containing the Azure CI report
+ const botComments = comments.data.filter(comment => comment.user.login === botUsername);
+ const lastComment = botComments.pop();
+
+ if (lastComment) {
+ const reportPrefix = '${lastCommitHash} Azure: '
+ const successReportString = '${reportPrefix}[SUCCESS]'
+ const failureReportString = '${reportPrefix}[FAILURE]'
+ if (lastComment.body.includes(reportPrefix)) {
+ if (lastComment.body.includes(successReportString)) {
+ console.log(`Azure CI succeeded on the latest commit of the PR.`);
+ return true;
+ } else if (lastComment.body.includes(failureReportString)) {
+ console.log(`Azure CI failed on the latest commit of the PR.`);
+ core.setFailed("Azure CI failed on the latest commit of the PR.");
+ return false;
+ } else {
+ console.log(`Azure CI is in progress on the latest commit of the PR.`);
+ core.setFailed("Azure CI is in progress on the latest commit of the PR.");
+ return false;
+ }
+ } else {
+ console.log(`No Azure CI report on the latest commit of the PR.`);
+ core.setFailed("No Azure CI report on the latest commit of the PR.");
+ return false;
+ }
+ } else {
+ console.log(`Azure CI report does not seem to be ready yet.`);
+ core.setFailed("Azure CI report does not seem to be ready yet.");
+ return false;
+ }
+ env:
+ GITHUB_TOKEN: ${{secrets.GITHUB_TOKEN}}
(hudi) 17/18: [HUDI-7443] Fix decimal conversion with legacy bytes type (#10756)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 5242b453236b0becc085881ef222cd8ffff6e44d
Author: stream2000 <st...@apache.org>
AuthorDate: Tue Feb 27 12:52:03 2024 +0800
[HUDI-7443] Fix decimal conversion with legacy bytes type (#10756)
---
.../org/apache/hudi/hadoop/utils/HiveAvroSerializer.java | 6 +++++-
.../hadoop/utils/HoodieRealtimeRecordReaderUtils.java | 16 +++++++++++-----
2 files changed, 16 insertions(+), 6 deletions(-)
diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HiveAvroSerializer.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HiveAvroSerializer.java
index 5f33844d60c..22116283d12 100644
--- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HiveAvroSerializer.java
+++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HiveAvroSerializer.java
@@ -289,7 +289,11 @@ public class HiveAvroSerializer {
HiveDecimal dec = (HiveDecimal) fieldOI.getPrimitiveJavaObject(structFieldData);
LogicalTypes.Decimal decimal = (LogicalTypes.Decimal) schema.getLogicalType();
BigDecimal bd = new BigDecimal(dec.toString()).setScale(decimal.getScale());
- return HoodieAvroUtils.DECIMAL_CONVERSION.toFixed(bd, schema, decimal);
+ if (schema.getType() == Schema.Type.BYTES) {
+ return HoodieAvroUtils.DECIMAL_CONVERSION.toBytes(bd, schema, decimal);
+ } else {
+ return HoodieAvroUtils.DECIMAL_CONVERSION.toFixed(bd, schema, decimal);
+ }
case CHAR:
HiveChar ch = (HiveChar) fieldOI.getPrimitiveJavaObject(structFieldData);
return new Utf8(ch.getStrippedValue());
diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeRecordReaderUtils.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeRecordReaderUtils.java
index 35fa7966c59..8ad61fc1704 100644
--- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeRecordReaderUtils.java
+++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeRecordReaderUtils.java
@@ -164,6 +164,9 @@ public class HoodieRealtimeRecordReaderUtils {
case STRING:
return new Text(value.toString());
case BYTES:
+ if (schema.getLogicalType() != null && schema.getLogicalType().getName().equals("decimal")) {
+ return toHiveDecimalWritable(((ByteBuffer) value).array(), schema);
+ }
return new BytesWritable(((ByteBuffer) value).array());
case INT:
if (schema.getLogicalType() != null && schema.getLogicalType().getName().equals("date")) {
@@ -245,11 +248,7 @@ public class HoodieRealtimeRecordReaderUtils {
}
case FIXED:
if (schema.getLogicalType() != null && schema.getLogicalType().getName().equals("decimal")) {
- LogicalTypes.Decimal decimal = (LogicalTypes.Decimal) LogicalTypes.fromSchema(schema);
- HiveDecimalWritable writable = new HiveDecimalWritable(((GenericFixed) value).bytes(),
- decimal.getScale());
- return HiveDecimalUtils.enforcePrecisionScale(writable,
- new DecimalTypeInfo(decimal.getPrecision(), decimal.getScale()));
+ return toHiveDecimalWritable(((GenericFixed) value).bytes(), schema);
}
return new BytesWritable(((GenericFixed) value).bytes());
default:
@@ -316,4 +315,11 @@ public class HoodieRealtimeRecordReaderUtils {
}
return appendFieldsToSchema(schema, newFields);
}
+
+ private static HiveDecimalWritable toHiveDecimalWritable(byte[] bytes, Schema schema) {
+ LogicalTypes.Decimal decimal = (LogicalTypes.Decimal) LogicalTypes.fromSchema(schema);
+ HiveDecimalWritable writable = new HiveDecimalWritable(bytes, decimal.getScale());
+ return HiveDecimalUtils.enforcePrecisionScale(writable,
+ new DecimalTypeInfo(decimal.getPrecision(), decimal.getScale()));
+ }
}
(hudi) 09/18: [HUDI-7438] Fix workflow condition for issue_comment events (#10749)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit d74c8cf432dc42fb59fe9e388c4c2ed9721e950b
Author: Y Ethan Guo <et...@gmail.com>
AuthorDate: Sat Feb 24 21:53:56 2024 -0800
[HUDI-7438] Fix workflow condition for issue_comment events (#10749)
---
.github/workflows/azure_ci_check.yml | 3 +--
1 file changed, 1 insertion(+), 2 deletions(-)
diff --git a/.github/workflows/azure_ci_check.yml b/.github/workflows/azure_ci_check.yml
index 17484a40aa5..1d10a23a520 100644
--- a/.github/workflows/azure_ci_check.yml
+++ b/.github/workflows/azure_ci_check.yml
@@ -30,8 +30,7 @@ jobs:
check-azure-ci-and-add-commit-status:
if: |
github.event.issue.pull_request != null &&
- github.event.issue.pull_request != '' &&
- github.event.issue_comment.user.login == 'hudi-bot'
+ github.event.comment.user.login == 'hudi-bot'
runs-on: ubuntu-latest
steps:
- name: Checkout repository
(hudi) 11/18: [MINOR] Fix typos in hudi-common module (#10748)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 09d311360475bf7498f4a675743c23d79342022b
Author: Y Ethan Guo <et...@gmail.com>
AuthorDate: Sun Feb 25 10:57:39 2024 -0800
[MINOR] Fix typos in hudi-common module (#10748)
---
hudi-common/src/main/java/org/apache/hudi/avro/AvroSchemaUtils.java | 2 +-
hudi-common/src/main/java/org/apache/hudi/common/bloom/Key.java | 2 +-
.../org/apache/hudi/common/util/queue/BaseHoodieQueueBasedExecutor.java | 2 +-
.../src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java | 2 +-
4 files changed, 4 insertions(+), 4 deletions(-)
diff --git a/hudi-common/src/main/java/org/apache/hudi/avro/AvroSchemaUtils.java b/hudi-common/src/main/java/org/apache/hudi/avro/AvroSchemaUtils.java
index 3c5486c47c7..5ec466cca3d 100644
--- a/hudi-common/src/main/java/org/apache/hudi/avro/AvroSchemaUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/avro/AvroSchemaUtils.java
@@ -57,7 +57,7 @@ public class AvroSchemaUtils {
/**
* Establishes whether {@code newSchema} is compatible w/ {@code prevSchema}, as
* defined by Avro's {@link AvroSchemaCompatibility}.
- * From avro's compatability standpoint, prevSchema is writer schema and new schema is reader schema.
+ * From avro's compatibility standpoint, prevSchema is writer schema and new schema is reader schema.
* {@code newSchema} is considered compatible to {@code prevSchema}, iff data written using {@code prevSchema}
* could be read by {@code newSchema}
*
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/bloom/Key.java b/hudi-common/src/main/java/org/apache/hudi/common/bloom/Key.java
index 37ae6e68f73..f14d301ae3b 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/bloom/Key.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/bloom/Key.java
@@ -136,7 +136,7 @@ public class Key implements Comparable<Key> {
/**
* Serialize the fields of this object to <code>out</code>.
*
- * @param out <code>DataOuput</code> to serialize this object into.
+ * @param out <code>DataOutput</code> to serialize this object into.
* @throws IOException
*/
public void write(DataOutput out) throws IOException {
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/util/queue/BaseHoodieQueueBasedExecutor.java b/hudi-common/src/main/java/org/apache/hudi/common/util/queue/BaseHoodieQueueBasedExecutor.java
index 20b9c802f60..f2843c56b03 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/util/queue/BaseHoodieQueueBasedExecutor.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/util/queue/BaseHoodieQueueBasedExecutor.java
@@ -215,7 +215,7 @@ public abstract class BaseHoodieQueueBasedExecutor<I, O, E> implements HoodieExe
// to be interrupted as well
Thread.currentThread().interrupt();
}
- // throw if we have any other exception seen already. There is a chance that cancellation/closing of producers with CompeletableFuture wins before the actual exception
+ // throw if we have any other exception seen already. There is a chance that cancellation/closing of producers with CompletableFuture wins before the actual exception
// is thrown.
if (this.queue.getThrowable() != null) {
throw new HoodieException(queue.getThrowable());
diff --git a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
index 7167a785f9f..0aa11042ab9 100644
--- a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
+++ b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
@@ -1615,7 +1615,7 @@ public class HoodieTableMetadataUtil {
*
* @param partitionType Type of the partition for which the file group count is to be estimated.
* @param recordCount The number of records expected to be written.
- * @param averageRecordSize Average size of each record to be writen.
+ * @param averageRecordSize Average size of each record to be written.
* @param minFileGroupCount Minimum number of file groups to use.
* @param maxFileGroupCount Maximum number of file groups to use.
* @param growthFactor By what factor are the records (recordCount) expected to grow?
(hudi) 10/18: [HUDI-7438] Fix issue number fetch in Azure CI check (#10751)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit a61c5c015b60624519e1f9a86045ad37a2d1ba72
Author: Y Ethan Guo <et...@gmail.com>
AuthorDate: Sun Feb 25 09:10:32 2024 -0800
[HUDI-7438] Fix issue number fetch in Azure CI check (#10751)
---
.github/workflows/azure_ci_check.yml | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/.github/workflows/azure_ci_check.yml b/.github/workflows/azure_ci_check.yml
index 1d10a23a520..1e33e6b8fa5 100644
--- a/.github/workflows/azure_ci_check.yml
+++ b/.github/workflows/azure_ci_check.yml
@@ -42,7 +42,7 @@ jobs:
with:
github-token: ${{secrets.GITHUB_TOKEN}}
script: |
- const issueNumber = github.event.issue.number;
+ const issueNumber = context.issue.number;
const { data: pullRequest } = await github.rest.pulls.get({
owner: context.repo.owner,
repo: context.repo.repo,
@@ -69,7 +69,7 @@ jobs:
github-token: ${{secrets.GITHUB_TOKEN}}
script: |
const latestCommitHash = '${{ steps.check_pr_state.outputs.latest_commit_hash }}'
- const issueNumber = github.event.issue.number;
+ const issueNumber = context.issue.number;
const checkAzureCiAndCreateCommitStatus = require(`${process.env.GITHUB_WORKSPACE}/.github/workflows/azure_ci.js`);
await checkAzureCiAndCreateCommitStatus({
(hudi) 01/18: [MINOR] Move release candidate validation to a separate GitHub action (#10729)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit d361e80c083e4c163c2b5ce3681c15b70b977c88
Author: Y Ethan Guo <et...@gmail.com>
AuthorDate: Thu Feb 22 17:00:41 2024 -0800
[MINOR] Move release candidate validation to a separate GitHub action (#10729)
---
.github/workflows/bot.yml | 72 ---------------
.github/workflows/release_candidate_validation.yml | 100 +++++++++++++++++++++
2 files changed, 100 insertions(+), 72 deletions(-)
diff --git a/.github/workflows/bot.yml b/.github/workflows/bot.yml
index a31c2e3ea35..ca53f8f6fdc 100644
--- a/.github/workflows/bot.yml
+++ b/.github/workflows/bot.yml
@@ -368,78 +368,6 @@ jobs:
HUDI_VERSION=$(mvn help:evaluate -Dexpression=project.version -q -DforceStdout)
./packaging/bundle-validation/ci_run.sh $HUDI_VERSION openjdk17
- validate-release-candidate-bundles:
- if: false
- runs-on: ubuntu-latest
- env:
- HUDI_VERSION: 0.13.1-rcx
- STAGING_REPO_NUM: 1123
- strategy:
- matrix:
- include:
- - flinkProfile: 'flink1.18'
- sparkProfile: 'spark3'
- sparkRuntime: 'spark3.5.0'
- - flinkProfile: 'flink1.18'
- sparkProfile: 'spark3.5'
- sparkRuntime: 'spark3.5.0'
- - flinkProfile: 'flink1.18'
- sparkProfile: 'spark3.4'
- sparkRuntime: 'spark3.4.0'
- - flinkProfile: 'flink1.17'
- sparkProfile: 'spark3.3'
- sparkRuntime: 'spark3.3.2'
- - flinkProfile: 'flink1.16'
- sparkProfile: 'spark3.3'
- sparkRuntime: 'spark3.3.1'
- - flinkProfile: 'flink1.15'
- sparkProfile: 'spark3.2'
- sparkRuntime: 'spark3.2.3'
- - flinkProfile: 'flink1.14'
- sparkProfile: 'spark3.1'
- sparkRuntime: 'spark3.1.3'
- - flinkProfile: 'flink1.14'
- sparkProfile: 'spark3.0'
- sparkRuntime: 'spark3.0.2'
- - flinkProfile: 'flink1.14'
- sparkProfile: 'spark'
- sparkRuntime: 'spark2.4.8'
- - flinkProfile: 'flink1.14'
- sparkProfile: 'spark2.4'
- sparkRuntime: 'spark2.4.8'
- steps:
- - uses: actions/checkout@v3
- - name: Set up JDK 8
- uses: actions/setup-java@v3
- with:
- java-version: '8'
- distribution: 'adopt'
- architecture: x64
- cache: maven
- - name: IT - Bundle Validation - OpenJDK 8
- env:
- FLINK_PROFILE: ${{ matrix.flinkProfile }}
- SPARK_PROFILE: ${{ matrix.sparkProfile }}
- SPARK_RUNTIME: ${{ matrix.sparkRuntime }}
- run: |
- ./packaging/bundle-validation/ci_run.sh $HUDI_VERSION openjdk8 $STAGING_REPO_NUM
- - name: IT - Bundle Validation - OpenJDK 11
- env:
- FLINK_PROFILE: ${{ matrix.flinkProfile }}
- SPARK_PROFILE: ${{ matrix.sparkProfile }}
- SPARK_RUNTIME: ${{ matrix.sparkRuntime }}
- if: ${{ startsWith(env.SPARK_PROFILE, 'spark3') }} # Only Spark 3.x supports Java 11 as of now
- run: |
- ./packaging/bundle-validation/ci_run.sh $HUDI_VERSION openjdk11 $STAGING_REPO_NUM
- - name: IT - Bundle Validation - OpenJDK 17
- env:
- FLINK_PROFILE: ${{ matrix.flinkProfile }}
- SPARK_PROFILE: ${{ matrix.sparkProfile }}
- SPARK_RUNTIME: ${{ matrix.sparkRuntime }}
- if: ${{ endsWith(env.SPARK_PROFILE, '3.3') }} # Only Spark 3.3 supports Java 17 as of now
- run: |
- ./packaging/bundle-validation/ci_run.sh $HUDI_VERSION openjdk17 $STAGING_REPO_NUM
-
integration-tests:
runs-on: ubuntu-latest
strategy:
diff --git a/.github/workflows/release_candidate_validation.yml b/.github/workflows/release_candidate_validation.yml
new file mode 100644
index 00000000000..2f14fd96f7d
--- /dev/null
+++ b/.github/workflows/release_candidate_validation.yml
@@ -0,0 +1,100 @@
+name: Release Candidate Validation
+
+on:
+ push:
+ branches:
+ - 'release-*'
+ pull_request:
+ paths-ignore:
+ - '**.bmp'
+ - '**.gif'
+ - '**.jpg'
+ - '**.jpeg'
+ - '**.md'
+ - '**.pdf'
+ - '**.png'
+ - '**.svg'
+ - '**.yaml'
+ - '.gitignore'
+ branches:
+ - 'release-*'
+
+concurrency:
+ group: ${{ github.ref }}
+ cancel-in-progress: ${{ !contains(github.ref, 'master') }}
+
+env:
+ MVN_ARGS: -e -ntp -B -V -Dgpg.skip -Djacoco.skip -Pwarn-log -Dorg.slf4j.simpleLogger.log.org.apache.maven.plugins.shade=warn -Dorg.slf4j.simpleLogger.log.org.apache.maven.plugins.dependency=warn -Dmaven.wagon.httpconnectionManager.ttlSeconds=25 -Dmaven.wagon.http.retryHandler.count=5
+ SPARK_COMMON_MODULES: hudi-spark-datasource/hudi-spark,hudi-spark-datasource/hudi-spark-common
+
+jobs:
+ validate-release-candidate-bundles:
+ runs-on: ubuntu-latest
+ env:
+ HUDI_VERSION: 0.14.1
+ STAGING_REPO_NUM: 1123
+ strategy:
+ matrix:
+ include:
+ - flinkProfile: 'flink1.18'
+ sparkProfile: 'spark3'
+ sparkRuntime: 'spark3.5.0'
+ - flinkProfile: 'flink1.18'
+ sparkProfile: 'spark3.5'
+ sparkRuntime: 'spark3.5.0'
+ - flinkProfile: 'flink1.18'
+ sparkProfile: 'spark3.4'
+ sparkRuntime: 'spark3.4.0'
+ - flinkProfile: 'flink1.17'
+ sparkProfile: 'spark3.3'
+ sparkRuntime: 'spark3.3.2'
+ - flinkProfile: 'flink1.16'
+ sparkProfile: 'spark3.3'
+ sparkRuntime: 'spark3.3.1'
+ - flinkProfile: 'flink1.15'
+ sparkProfile: 'spark3.2'
+ sparkRuntime: 'spark3.2.3'
+ - flinkProfile: 'flink1.14'
+ sparkProfile: 'spark3.1'
+ sparkRuntime: 'spark3.1.3'
+ - flinkProfile: 'flink1.14'
+ sparkProfile: 'spark3.0'
+ sparkRuntime: 'spark3.0.2'
+ - flinkProfile: 'flink1.14'
+ sparkProfile: 'spark'
+ sparkRuntime: 'spark2.4.8'
+ - flinkProfile: 'flink1.14'
+ sparkProfile: 'spark2.4'
+ sparkRuntime: 'spark2.4.8'
+ steps:
+ - uses: actions/checkout@v3
+ - name: Set up JDK 8
+ uses: actions/setup-java@v3
+ with:
+ java-version: '8'
+ distribution: 'adopt'
+ architecture: x64
+ cache: maven
+ - name: IT - Bundle Validation - OpenJDK 8
+ env:
+ FLINK_PROFILE: ${{ matrix.flinkProfile }}
+ SPARK_PROFILE: ${{ matrix.sparkProfile }}
+ SPARK_RUNTIME: ${{ matrix.sparkRuntime }}
+ run: |
+ ./packaging/bundle-validation/ci_run.sh $HUDI_VERSION openjdk8 $STAGING_REPO_NUM
+ - name: IT - Bundle Validation - OpenJDK 11
+ env:
+ FLINK_PROFILE: ${{ matrix.flinkProfile }}
+ SPARK_PROFILE: ${{ matrix.sparkProfile }}
+ SPARK_RUNTIME: ${{ matrix.sparkRuntime }}
+ if: ${{ startsWith(env.SPARK_PROFILE, 'spark3') }} # Only Spark 3.x supports Java 11 as of now
+ run: |
+ ./packaging/bundle-validation/ci_run.sh $HUDI_VERSION openjdk11 $STAGING_REPO_NUM
+ - name: IT - Bundle Validation - OpenJDK 17
+ env:
+ FLINK_PROFILE: ${{ matrix.flinkProfile }}
+ SPARK_PROFILE: ${{ matrix.sparkProfile }}
+ SPARK_RUNTIME: ${{ matrix.sparkRuntime }}
+ if: ${{ endsWith(env.SPARK_PROFILE, '3.3') }} # Only Spark 3.3 supports Java 17 as of now
+ run: |
+ ./packaging/bundle-validation/ci_run.sh $HUDI_VERSION openjdk17 $STAGING_REPO_NUM
(hudi) 16/18: [MINOR][DOCS] Update comment on hoodiemultitablestreamer (#10667)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 413324346a77ed419ebbd1ab7f56162984942287
Author: nadine farah <nf...@gmail.com>
AuthorDate: Mon Feb 26 16:55:06 2024 -0800
[MINOR][DOCS] Update comment on hoodiemultitablestreamer (#10667)
---
.../org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java
index d7e3bca4989..a637f7fbbff 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java
@@ -66,7 +66,7 @@ import static org.apache.hudi.utilities.config.HoodieStreamerConfig.TRANSFORMER_
/**
* Wrapper over HoodieStreamer.java class.
* Helps with ingesting incremental data into hoodie datasets for multiple tables.
- * Currently supports only COPY_ON_WRITE storage type.
+ * Supports COPY_ON_WRITE and MERGE_ON_READ storage types.
*/
public class HoodieMultiTableStreamer {
(hudi) 13/18: [HUDI-7416] Add interface for SourceProfile to be used in StreamSync for reading data (#10736)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 7eb05e2d89d3d1572e1b15716747ef64ec190d58
Author: Vinish Reddy <vi...@gmail.com>
AuthorDate: Mon Feb 26 20:44:50 2024 +0530
[HUDI-7416] Add interface for SourceProfile to be used in StreamSync for reading data (#10736)
Introducing a new class known as SourceProfile which contains details about how the next sync round in StreamSync should be consumed. For eg:
KafkaSourceProfile contains number of events to consume in this sync round.
S3SourceProfile contains the list of files to consume in this sync round
HudiIncrementalSourceProfile contains the beginInstant and endInstant commit times to consume in this sync round.
In future we can add the method for choosing the writeOperationType and indexType as well, for sourceProfile.getSourceSpecificContext() will be used to consume the data from the source.
---
.../org/apache/hudi/utilities/UtilHelpers.java | 19 ++++++++
.../hudi/utilities/deltastreamer/DeltaSync.java | 4 +-
.../hudi/utilities/sources/AvroKafkaSource.java | 13 ++++--
.../hudi/utilities/sources/JsonKafkaSource.java | 14 ++++--
.../apache/hudi/utilities/sources/KafkaSource.java | 39 +++++++++++-----
.../hudi/utilities/sources/ProtoKafkaSource.java | 15 ++++--
.../org/apache/hudi/utilities/sources/Source.java | 11 ++++-
.../utilities/sources/helpers/KafkaOffsetGen.java | 39 ++++++++--------
.../utilities/streamer/DefaultStreamContext.java | 48 +++++++++++++++++++
.../hudi/utilities/streamer/HoodieStreamer.java | 21 ++++++---
.../hudi/utilities/streamer/SourceProfile.java | 54 ++++++++++++++++++++++
.../utilities/streamer/SourceProfileSupplier.java | 34 ++++++++++++++
.../hudi/utilities/streamer/StreamContext.java | 44 ++++++++++++++++++
.../apache/hudi/utilities/streamer/StreamSync.java | 10 ++--
.../utilities/sources/BaseTestKafkaSource.java | 51 ++++++++++++++++++++
.../utilities/sources/TestJsonKafkaSource.java | 15 +++++-
.../utilities/sources/TestProtoKafkaSource.java | 3 +-
17 files changed, 374 insertions(+), 60 deletions(-)
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/UtilHelpers.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/UtilHelpers.java
index 3b789bae022..d0781849755 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/UtilHelpers.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/UtilHelpers.java
@@ -66,6 +66,7 @@ import org.apache.hudi.utilities.sources.InputBatch;
import org.apache.hudi.utilities.sources.Source;
import org.apache.hudi.utilities.sources.processor.ChainedJsonKafkaSourcePostProcessor;
import org.apache.hudi.utilities.sources.processor.JsonKafkaSourcePostProcessor;
+import org.apache.hudi.utilities.streamer.StreamContext;
import org.apache.hudi.utilities.transform.ChainedTransformer;
import org.apache.hudi.utilities.transform.ErrorTableAwareChainedTransformer;
import org.apache.hudi.utilities.transform.Transformer;
@@ -156,6 +157,24 @@ public class UtilHelpers {
}
}
+ public static Source createSource(String sourceClass, TypedProperties cfg, JavaSparkContext jssc,
+ SparkSession sparkSession, HoodieIngestionMetrics metrics, StreamContext streamContext)
+ throws IOException {
+ try {
+ try {
+ return (Source) ReflectionUtils.loadClass(sourceClass,
+ new Class<?>[] {TypedProperties.class, JavaSparkContext.class,
+ SparkSession.class,
+ HoodieIngestionMetrics.class, StreamContext.class},
+ cfg, jssc, sparkSession, metrics, streamContext);
+ } catch (HoodieException e) {
+ return createSource(sourceClass, cfg, jssc, sparkSession, streamContext.getSchemaProvider(), metrics);
+ }
+ } catch (Throwable e) {
+ throw new IOException("Could not load source class " + sourceClass, e);
+ }
+ }
+
public static JsonKafkaSourcePostProcessor createJsonKafkaSourcePostProcessor(String postProcessorClassNames, TypedProperties props) throws IOException {
if (StringUtils.isNullOrEmpty(postProcessorClassNames)) {
return null;
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
index c794db32510..4002d1579bb 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
@@ -22,7 +22,9 @@ package org.apache.hudi.utilities.deltastreamer;
import org.apache.hudi.client.SparkRDDWriteClient;
import org.apache.hudi.client.common.HoodieSparkEngineContext;
import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.util.Option;
import org.apache.hudi.utilities.schema.SchemaProvider;
+import org.apache.hudi.utilities.streamer.DefaultStreamContext;
import org.apache.hudi.utilities.streamer.HoodieStreamer;
import org.apache.hudi.utilities.streamer.StreamSync;
@@ -49,6 +51,6 @@ public class DeltaSync extends StreamSync {
public DeltaSync(HoodieDeltaStreamer.Config cfg, SparkSession sparkSession, SchemaProvider schemaProvider,
TypedProperties props, HoodieSparkEngineContext hoodieSparkContext, FileSystem fs, Configuration conf,
Function<SparkRDDWriteClient, Boolean> onInitializingHoodieWriteClient) throws IOException {
- super(cfg, sparkSession, schemaProvider, props, hoodieSparkContext, fs, conf, onInitializingHoodieWriteClient);
+ super(cfg, sparkSession, props, hoodieSparkContext, fs, conf, onInitializingHoodieWriteClient, new DefaultStreamContext(schemaProvider, Option.empty()));
}
}
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/AvroKafkaSource.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/AvroKafkaSource.java
index 2bf92280faf..36c83d63030 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/AvroKafkaSource.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/AvroKafkaSource.java
@@ -27,6 +27,8 @@ import org.apache.hudi.utilities.ingestion.HoodieIngestionMetrics;
import org.apache.hudi.utilities.schema.SchemaProvider;
import org.apache.hudi.utilities.sources.helpers.AvroConvertor;
import org.apache.hudi.utilities.sources.helpers.KafkaOffsetGen;
+import org.apache.hudi.utilities.streamer.DefaultStreamContext;
+import org.apache.hudi.utilities.streamer.StreamContext;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -69,10 +71,13 @@ public class AvroKafkaSource extends KafkaSource<GenericRecord> {
public AvroKafkaSource(TypedProperties props, JavaSparkContext sparkContext, SparkSession sparkSession,
SchemaProvider schemaProvider, HoodieIngestionMetrics metrics) {
- super(props, sparkContext, sparkSession,
- UtilHelpers.getSchemaProviderForKafkaSource(schemaProvider, props, sparkContext),
- SourceType.AVRO, metrics);
- this.originalSchemaProvider = schemaProvider;
+ this(props, sparkContext, sparkSession, metrics, new DefaultStreamContext(schemaProvider, Option.empty()));
+ }
+
+ public AvroKafkaSource(TypedProperties properties, JavaSparkContext sparkContext, SparkSession sparkSession, HoodieIngestionMetrics metrics, StreamContext streamContext) {
+ super(properties, sparkContext, sparkSession, SourceType.AVRO, metrics,
+ new DefaultStreamContext(UtilHelpers.getSchemaProviderForKafkaSource(streamContext.getSchemaProvider(), properties, sparkContext), streamContext.getSourceProfileSupplier()));
+ this.originalSchemaProvider = streamContext.getSchemaProvider();
props.put(NATIVE_KAFKA_KEY_DESERIALIZER_PROP, StringDeserializer.class.getName());
deserializerClassName = getStringWithAltKeys(props, KAFKA_AVRO_VALUE_DESERIALIZER_CLASS, true);
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/JsonKafkaSource.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/JsonKafkaSource.java
index eb67abfee3a..6e95a315260 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/JsonKafkaSource.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/JsonKafkaSource.java
@@ -19,6 +19,7 @@
package org.apache.hudi.utilities.sources;
import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.utilities.UtilHelpers;
import org.apache.hudi.utilities.config.JsonKafkaPostProcessorConfig;
@@ -27,6 +28,8 @@ import org.apache.hudi.utilities.ingestion.HoodieIngestionMetrics;
import org.apache.hudi.utilities.schema.SchemaProvider;
import org.apache.hudi.utilities.sources.helpers.KafkaOffsetGen;
import org.apache.hudi.utilities.sources.processor.JsonKafkaSourcePostProcessor;
+import org.apache.hudi.utilities.streamer.DefaultStreamContext;
+import org.apache.hudi.utilities.streamer.StreamContext;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
@@ -44,10 +47,10 @@ import java.util.LinkedList;
import java.util.List;
import static org.apache.hudi.common.util.ConfigUtils.getStringWithAltKeys;
+import static org.apache.hudi.utilities.schema.KafkaOffsetPostProcessor.KAFKA_SOURCE_KEY_COLUMN;
import static org.apache.hudi.utilities.schema.KafkaOffsetPostProcessor.KAFKA_SOURCE_OFFSET_COLUMN;
import static org.apache.hudi.utilities.schema.KafkaOffsetPostProcessor.KAFKA_SOURCE_PARTITION_COLUMN;
import static org.apache.hudi.utilities.schema.KafkaOffsetPostProcessor.KAFKA_SOURCE_TIMESTAMP_COLUMN;
-import static org.apache.hudi.utilities.schema.KafkaOffsetPostProcessor.KAFKA_SOURCE_KEY_COLUMN;
/**
* Read json kafka data.
@@ -56,9 +59,12 @@ public class JsonKafkaSource extends KafkaSource<String> {
public JsonKafkaSource(TypedProperties properties, JavaSparkContext sparkContext, SparkSession sparkSession,
SchemaProvider schemaProvider, HoodieIngestionMetrics metrics) {
- super(properties, sparkContext, sparkSession,
- UtilHelpers.getSchemaProviderForKafkaSource(schemaProvider, properties, sparkContext),
- SourceType.JSON, metrics);
+ this(properties, sparkContext, sparkSession, metrics, new DefaultStreamContext(schemaProvider, Option.empty()));
+ }
+
+ public JsonKafkaSource(TypedProperties properties, JavaSparkContext sparkContext, SparkSession sparkSession, HoodieIngestionMetrics metrics, StreamContext streamContext) {
+ super(properties, sparkContext, sparkSession, SourceType.JSON, metrics,
+ new DefaultStreamContext(UtilHelpers.getSchemaProviderForKafkaSource(streamContext.getSchemaProvider(), properties, sparkContext), streamContext.getSourceProfileSupplier()));
properties.put("key.deserializer", StringDeserializer.class.getName());
properties.put("value.deserializer", StringDeserializer.class.getName());
this.offsetGen = new KafkaOffsetGen(props);
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/KafkaSource.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/KafkaSource.java
index bb26d579582..52a6a1217cc 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/KafkaSource.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/KafkaSource.java
@@ -26,6 +26,8 @@ import org.apache.hudi.utilities.ingestion.HoodieIngestionMetrics;
import org.apache.hudi.utilities.schema.KafkaOffsetPostProcessor;
import org.apache.hudi.utilities.schema.SchemaProvider;
import org.apache.hudi.utilities.sources.helpers.KafkaOffsetGen;
+import org.apache.hudi.utilities.streamer.SourceProfile;
+import org.apache.hudi.utilities.streamer.StreamContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
@@ -50,9 +52,9 @@ abstract class KafkaSource<T> extends Source<JavaRDD<T>> {
protected final boolean shouldAddOffsets;
protected KafkaSource(TypedProperties props, JavaSparkContext sparkContext, SparkSession sparkSession,
- SchemaProvider schemaProvider, SourceType sourceType, HoodieIngestionMetrics metrics) {
- super(props, sparkContext, sparkSession, schemaProvider, sourceType);
- this.schemaProvider = schemaProvider;
+ SourceType sourceType, HoodieIngestionMetrics metrics, StreamContext streamContext) {
+ super(props, sparkContext, sparkSession, sourceType, streamContext);
+ this.schemaProvider = streamContext.getSchemaProvider();
this.metrics = metrics;
this.shouldAddOffsets = KafkaOffsetPostProcessor.Config.shouldAddOffsets(props);
}
@@ -60,21 +62,34 @@ abstract class KafkaSource<T> extends Source<JavaRDD<T>> {
@Override
protected InputBatch<JavaRDD<T>> fetchNewData(Option<String> lastCheckpointStr, long sourceLimit) {
try {
- OffsetRange[] offsetRanges = offsetGen.getNextOffsetRanges(lastCheckpointStr, sourceLimit, metrics);
- long totalNewMsgs = KafkaOffsetGen.CheckpointUtils.totalNewMessages(offsetRanges);
- LOG.info("About to read " + totalNewMsgs + " from Kafka for topic :" + offsetGen.getTopicName());
- if (totalNewMsgs <= 0) {
- metrics.updateStreamerSourceNewMessageCount(METRIC_NAME_KAFKA_MESSAGE_IN_COUNT, 0);
- return new InputBatch<>(Option.empty(), KafkaOffsetGen.CheckpointUtils.offsetsToStr(offsetRanges));
+ OffsetRange[] offsetRanges;
+ if (sourceProfileSupplier.isPresent() && sourceProfileSupplier.get().getSourceProfile() != null) {
+ SourceProfile<Long> kafkaSourceProfile = sourceProfileSupplier.get().getSourceProfile();
+ offsetRanges = offsetGen.getNextOffsetRanges(lastCheckpointStr, kafkaSourceProfile.getSourceSpecificContext(), kafkaSourceProfile.getSourcePartitions(), metrics);
+ LOG.info("About to read numEvents {} of size {} bytes in {} partitions from Kafka for topic {} with offsetRanges {}",
+ kafkaSourceProfile.getSourceSpecificContext(), kafkaSourceProfile.getMaxSourceBytes(),
+ kafkaSourceProfile.getSourcePartitions(), offsetGen.getTopicName(), offsetRanges);
+ } else {
+ offsetRanges = offsetGen.getNextOffsetRanges(lastCheckpointStr, sourceLimit, metrics);
}
- metrics.updateStreamerSourceNewMessageCount(METRIC_NAME_KAFKA_MESSAGE_IN_COUNT, totalNewMsgs);
- JavaRDD<T> newDataRDD = toRDD(offsetRanges);
- return new InputBatch<>(Option.of(newDataRDD), KafkaOffsetGen.CheckpointUtils.offsetsToStr(offsetRanges));
+ return toInputBatch(offsetRanges);
} catch (org.apache.kafka.common.errors.TimeoutException e) {
throw new HoodieSourceTimeoutException("Kafka Source timed out " + e.getMessage());
}
}
+ private InputBatch<JavaRDD<T>> toInputBatch(OffsetRange[] offsetRanges) {
+ long totalNewMsgs = KafkaOffsetGen.CheckpointUtils.totalNewMessages(offsetRanges);
+ LOG.info("About to read " + totalNewMsgs + " from Kafka for topic :" + offsetGen.getTopicName());
+ if (totalNewMsgs <= 0) {
+ metrics.updateStreamerSourceNewMessageCount(METRIC_NAME_KAFKA_MESSAGE_IN_COUNT, 0);
+ return new InputBatch<>(Option.empty(), KafkaOffsetGen.CheckpointUtils.offsetsToStr(offsetRanges));
+ }
+ metrics.updateStreamerSourceNewMessageCount(METRIC_NAME_KAFKA_MESSAGE_IN_COUNT, totalNewMsgs);
+ JavaRDD<T> newDataRDD = toRDD(offsetRanges);
+ return new InputBatch<>(Option.of(newDataRDD), KafkaOffsetGen.CheckpointUtils.offsetsToStr(offsetRanges));
+ }
+
abstract JavaRDD<T> toRDD(OffsetRange[] offsetRanges);
@Override
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/ProtoKafkaSource.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/ProtoKafkaSource.java
index 67927480454..d7a15b3932c 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/ProtoKafkaSource.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/ProtoKafkaSource.java
@@ -19,12 +19,16 @@
package org.apache.hudi.utilities.sources;
import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ReflectionUtils;
+import org.apache.hudi.utilities.UtilHelpers;
import org.apache.hudi.utilities.config.ProtoClassBasedSchemaProviderConfig;
import org.apache.hudi.utilities.exception.HoodieReadFromSourceException;
import org.apache.hudi.utilities.ingestion.HoodieIngestionMetrics;
import org.apache.hudi.utilities.schema.SchemaProvider;
import org.apache.hudi.utilities.sources.helpers.KafkaOffsetGen;
+import org.apache.hudi.utilities.streamer.DefaultStreamContext;
+import org.apache.hudi.utilities.streamer.StreamContext;
import com.google.protobuf.Message;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
@@ -51,9 +55,14 @@ public class ProtoKafkaSource extends KafkaSource<Message> {
private final String className;
- public ProtoKafkaSource(TypedProperties props, JavaSparkContext sparkContext,
- SparkSession sparkSession, SchemaProvider schemaProvider, HoodieIngestionMetrics metrics) {
- super(props, sparkContext, sparkSession, schemaProvider, SourceType.PROTO, metrics);
+ public ProtoKafkaSource(TypedProperties props, JavaSparkContext sparkContext, SparkSession sparkSession,
+ SchemaProvider schemaProvider, HoodieIngestionMetrics metrics) {
+ this(props, sparkContext, sparkSession, metrics, new DefaultStreamContext(schemaProvider, Option.empty()));
+ }
+
+ public ProtoKafkaSource(TypedProperties properties, JavaSparkContext sparkContext, SparkSession sparkSession, HoodieIngestionMetrics metrics, StreamContext streamContext) {
+ super(properties, sparkContext, sparkSession, SourceType.PROTO, metrics,
+ new DefaultStreamContext(UtilHelpers.getSchemaProviderForKafkaSource(streamContext.getSchemaProvider(), properties, sparkContext), streamContext.getSourceProfileSupplier()));
checkRequiredConfigProperties(props, Collections.singletonList(
ProtoClassBasedSchemaProviderConfig.PROTO_SCHEMA_CLASS_NAME));
props.put(NATIVE_KAFKA_KEY_DESERIALIZER_PROP, StringDeserializer.class);
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/Source.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/Source.java
index cbc0722056b..dfb07c718a0 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/Source.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/Source.java
@@ -25,6 +25,9 @@ import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.utilities.callback.SourceCommitCallback;
import org.apache.hudi.utilities.schema.SchemaProvider;
+import org.apache.hudi.utilities.streamer.DefaultStreamContext;
+import org.apache.hudi.utilities.streamer.SourceProfileSupplier;
+import org.apache.hudi.utilities.streamer.StreamContext;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
@@ -44,6 +47,7 @@ public abstract class Source<T> implements SourceCommitCallback, Serializable {
protected transient TypedProperties props;
protected transient JavaSparkContext sparkContext;
protected transient SparkSession sparkSession;
+ protected transient Option<SourceProfileSupplier> sourceProfileSupplier;
private transient SchemaProvider overriddenSchemaProvider;
private final SourceType sourceType;
@@ -55,11 +59,16 @@ public abstract class Source<T> implements SourceCommitCallback, Serializable {
protected Source(TypedProperties props, JavaSparkContext sparkContext, SparkSession sparkSession,
SchemaProvider schemaProvider, SourceType sourceType) {
+ this(props, sparkContext, sparkSession, sourceType, new DefaultStreamContext(schemaProvider, Option.empty()));
+ }
+
+ protected Source(TypedProperties props, JavaSparkContext sparkContext, SparkSession sparkSession, SourceType sourceType, StreamContext streamContext) {
this.props = props;
this.sparkContext = sparkContext;
this.sparkSession = sparkSession;
- this.overriddenSchemaProvider = schemaProvider;
+ this.overriddenSchemaProvider = streamContext.getSchemaProvider();
this.sourceType = sourceType;
+ this.sourceProfileSupplier = streamContext.getSourceProfileSupplier();
}
@PublicAPIMethod(maturity = ApiMaturityLevel.STABLE)
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java
index d5faec3595e..32df651d556 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java
@@ -241,7 +241,24 @@ public class KafkaOffsetGen {
}
public OffsetRange[] getNextOffsetRanges(Option<String> lastCheckpointStr, long sourceLimit, HoodieIngestionMetrics metrics) {
+ // Come up with final set of OffsetRanges to read (account for new partitions, limit number of events)
+ long maxEventsToReadFromKafka = getLongWithAltKeys(props, KafkaSourceConfig.MAX_EVENTS_FROM_KAFKA_SOURCE);
+ long numEvents;
+ if (sourceLimit == Long.MAX_VALUE) {
+ numEvents = maxEventsToReadFromKafka;
+ LOG.info("SourceLimit not configured, set numEvents to default value : " + maxEventsToReadFromKafka);
+ } else {
+ numEvents = sourceLimit;
+ }
+
+ long minPartitions = getLongWithAltKeys(props, KafkaSourceConfig.KAFKA_SOURCE_MIN_PARTITIONS);
+ LOG.info("getNextOffsetRanges set config " + KafkaSourceConfig.KAFKA_SOURCE_MIN_PARTITIONS.key() + " to " + minPartitions);
+
+ return getNextOffsetRanges(lastCheckpointStr, numEvents, minPartitions, metrics);
+ }
+
+ public OffsetRange[] getNextOffsetRanges(Option<String> lastCheckpointStr, long numEvents, long minPartitions, HoodieIngestionMetrics metrics) {
// Obtain current metadata for the topic
Map<TopicPartition, Long> fromOffsets;
Map<TopicPartition, Long> toOffsets;
@@ -279,29 +296,9 @@ public class KafkaOffsetGen {
// Obtain the latest offsets.
toOffsets = consumer.endOffsets(topicPartitions);
}
-
- // Come up with final set of OffsetRanges to read (account for new partitions, limit number of events)
- long maxEventsToReadFromKafka = getLongWithAltKeys(props, KafkaSourceConfig.MAX_EVENTS_FROM_KAFKA_SOURCE);
-
- long numEvents;
- if (sourceLimit == Long.MAX_VALUE) {
- numEvents = maxEventsToReadFromKafka;
- LOG.info("SourceLimit not configured, set numEvents to default value : " + maxEventsToReadFromKafka);
- } else {
- numEvents = sourceLimit;
- }
-
- // TODO(HUDI-4625) remove
- if (numEvents < toOffsets.size()) {
- throw new HoodieException("sourceLimit should not be less than the number of kafka partitions");
- }
-
- long minPartitions = getLongWithAltKeys(props, KafkaSourceConfig.KAFKA_SOURCE_MIN_PARTITIONS);
- LOG.info("getNextOffsetRanges set config " + KafkaSourceConfig.KAFKA_SOURCE_MIN_PARTITIONS.key() + " to " + minPartitions);
-
return CheckpointUtils.computeOffsetRanges(fromOffsets, toOffsets, numEvents, minPartitions);
}
-
+
/**
* Fetch partition infos for given topic.
*
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/DefaultStreamContext.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/DefaultStreamContext.java
new file mode 100644
index 00000000000..f8dabeb89c9
--- /dev/null
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/DefaultStreamContext.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.utilities.streamer;
+
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.utilities.schema.SchemaProvider;
+
+/**
+ * The default implementation for the StreamContext interface,
+ * composes SchemaProvider and SourceProfileSupplier currently,
+ * can be extended for other arguments in the future.
+ */
+public class DefaultStreamContext implements StreamContext {
+
+ private final SchemaProvider schemaProvider;
+ private final Option<SourceProfileSupplier> sourceProfileSupplier;
+
+ public DefaultStreamContext(SchemaProvider schemaProvider, Option<SourceProfileSupplier> sourceProfileSupplier) {
+ this.schemaProvider = schemaProvider;
+ this.sourceProfileSupplier = sourceProfileSupplier;
+ }
+
+ @Override
+ public SchemaProvider getSchemaProvider() {
+ return schemaProvider;
+ }
+
+ @Override
+ public Option<SourceProfileSupplier> getSourceProfileSupplier() {
+ return sourceProfileSupplier;
+ }
+}
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java
index 9ff666b049c..72e5e1c36ef 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java
@@ -143,8 +143,12 @@ public class HoodieStreamer implements Serializable {
this(cfg, jssc, fs, conf, Option.empty());
}
+ public HoodieStreamer(Config cfg, JavaSparkContext jssc, FileSystem fs, Configuration conf, Option<TypedProperties> propsOverride) throws IOException {
+ this(cfg, jssc, fs, conf, propsOverride, Option.empty());
+ }
+
public HoodieStreamer(Config cfg, JavaSparkContext jssc, FileSystem fs, Configuration conf,
- Option<TypedProperties> propsOverride) throws IOException {
+ Option<TypedProperties> propsOverride, Option<SourceProfileSupplier> sourceProfileSupplier) throws IOException {
this.properties = combineProperties(cfg, propsOverride, jssc.hadoopConfiguration());
if (cfg.initialCheckpointProvider != null && cfg.checkpoint == null) {
InitialCheckPointProvider checkPointProvider =
@@ -158,7 +162,7 @@ public class HoodieStreamer implements Serializable {
cfg.runBootstrap ? new BootstrapExecutor(cfg, jssc, fs, conf, this.properties) : null);
HoodieSparkEngineContext sparkEngineContext = new HoodieSparkEngineContext(jssc);
this.ingestionService = Option.ofNullable(
- cfg.runBootstrap ? null : new StreamSyncService(cfg, sparkEngineContext, fs, conf, Option.ofNullable(this.properties)));
+ cfg.runBootstrap ? null : new StreamSyncService(cfg, sparkEngineContext, fs, conf, Option.ofNullable(this.properties), sourceProfileSupplier));
}
private static TypedProperties combineProperties(Config cfg, Option<TypedProperties> propsOverride, Configuration hadoopConf) {
@@ -656,7 +660,7 @@ public class HoodieStreamer implements Serializable {
private final Option<ConfigurationHotUpdateStrategy> configurationHotUpdateStrategyOpt;
public StreamSyncService(Config cfg, HoodieSparkEngineContext hoodieSparkContext, FileSystem fs, Configuration conf,
- Option<TypedProperties> properties) throws IOException {
+ Option<TypedProperties> properties, Option<SourceProfileSupplier> sourceProfileSupplier) throws IOException {
super(HoodieIngestionConfig.newBuilder()
.isContinuous(cfg.continuousMode)
.withMinSyncInternalSeconds(cfg.minSyncIntervalSeconds).build());
@@ -712,13 +716,18 @@ public class HoodieStreamer implements Serializable {
UtilHelpers.createSchemaProvider(cfg.schemaProviderClassName, props, hoodieSparkContext.jsc()),
props, hoodieSparkContext.jsc(), cfg.transformerClassNames);
- streamSync = new StreamSync(cfg, sparkSession, schemaProvider, props, hoodieSparkContext, fs, conf, this::onInitializingWriteClient);
+ streamSync = new StreamSync(cfg, sparkSession, props, hoodieSparkContext, fs, conf, this::onInitializingWriteClient, new DefaultStreamContext(schemaProvider, sourceProfileSupplier));
}
public StreamSyncService(HoodieStreamer.Config cfg, HoodieSparkEngineContext hoodieSparkContext, FileSystem fs, Configuration conf)
throws IOException {
- this(cfg, hoodieSparkContext, fs, conf, Option.empty());
+ this(cfg, hoodieSparkContext, fs, conf, Option.empty(), Option.empty());
+ }
+
+ public StreamSyncService(HoodieStreamer.Config cfg, HoodieSparkEngineContext hoodieSparkContext, FileSystem fs, Configuration conf, Option<TypedProperties> properties)
+ throws IOException {
+ this(cfg, hoodieSparkContext, fs, conf, properties, Option.empty());
}
private void initializeTableTypeAndBaseFileFormat() {
@@ -732,7 +741,7 @@ public class HoodieStreamer implements Serializable {
if (streamSync != null) {
streamSync.close();
}
- streamSync = new StreamSync(cfg, sparkSession, schemaProvider, props, hoodieSparkContext, fs, hiveConf, this::onInitializingWriteClient);
+ streamSync = new StreamSync(cfg, sparkSession, props, hoodieSparkContext, fs, hiveConf, this::onInitializingWriteClient, new DefaultStreamContext(schemaProvider, Option.empty()));
}
@Override
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/SourceProfile.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/SourceProfile.java
new file mode 100644
index 00000000000..d830cf5dee3
--- /dev/null
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/SourceProfile.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.utilities.streamer;
+
+import org.apache.hudi.ApiMaturityLevel;
+import org.apache.hudi.PublicAPIClass;
+import org.apache.hudi.PublicAPIMethod;
+
+/**
+ * A profile containing details about how the next input batch in StreamSync should be consumed and written.
+ * For eg: KafkaSourceProfile contains number of events to consume in this sync round.
+ * S3SourceProfile contains the list of files to consume in this sync round.
+ * HudiIncrementalSourceProfile contains the beginInstant and endInstant commit times to consume in this sync round etc.
+ *
+ * @param <T> The type for source context, varies based on sourceType as described above.
+ */
+@PublicAPIClass(maturity = ApiMaturityLevel.EVOLVING)
+public interface SourceProfile<T> {
+
+ /**
+ * @return The maxBytes that will be consumed from the source in this sync round.
+ */
+ @PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
+ long getMaxSourceBytes();
+
+ /**
+ * @return The number of output partitions required in source RDD.
+ */
+ @PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
+ int getSourcePartitions();
+
+ /**
+ * @return The source specific context based on sourceType as described above.
+ */
+ @PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
+ T getSourceSpecificContext();
+}
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/SourceProfileSupplier.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/SourceProfileSupplier.java
new file mode 100644
index 00000000000..34bfb8dff94
--- /dev/null
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/SourceProfileSupplier.java
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.utilities.streamer;
+
+import org.apache.hudi.ApiMaturityLevel;
+import org.apache.hudi.PublicAPIClass;
+import org.apache.hudi.PublicAPIMethod;
+
+/**
+ * Supplier for SourceProfile
+ */
+@PublicAPIClass(maturity = ApiMaturityLevel.EVOLVING)
+public interface SourceProfileSupplier {
+ @SuppressWarnings("rawtypes")
+ @PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
+ SourceProfile getSourceProfile();
+}
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamContext.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamContext.java
new file mode 100644
index 00000000000..bfe337ee3f2
--- /dev/null
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamContext.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.utilities.streamer;
+
+import org.apache.hudi.ApiMaturityLevel;
+import org.apache.hudi.PublicAPIClass;
+import org.apache.hudi.PublicAPIMethod;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.utilities.schema.SchemaProvider;
+
+/**
+ * The context required to sync one batch of data to hoodie table using StreamSync.
+ */
+@PublicAPIClass(maturity = ApiMaturityLevel.EVOLVING)
+public interface StreamContext {
+
+ /**
+ * The schema provider used for reading data from source and also writing to hoodie table.
+ */
+ @PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
+ SchemaProvider getSchemaProvider();
+
+ /**
+ * An optional stream profile supplying details regarding how the next input batch in StreamSync should be consumed and written.
+ */
+ @PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
+ Option<SourceProfileSupplier> getSourceProfileSupplier();
+}
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java
index 4c71abc66bc..fe8eb909db4 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java
@@ -259,19 +259,19 @@ public class StreamSync implements Serializable, Closeable {
public StreamSync(HoodieStreamer.Config cfg, SparkSession sparkSession, SchemaProvider schemaProvider,
TypedProperties props, JavaSparkContext jssc, FileSystem fs, Configuration conf,
Function<SparkRDDWriteClient, Boolean> onInitializingHoodieWriteClient) throws IOException {
- this(cfg, sparkSession, schemaProvider, props, new HoodieSparkEngineContext(jssc), fs, conf, onInitializingHoodieWriteClient);
+ this(cfg, sparkSession, props, new HoodieSparkEngineContext(jssc), fs, conf, onInitializingHoodieWriteClient, new DefaultStreamContext(schemaProvider, Option.empty()));
}
- public StreamSync(HoodieStreamer.Config cfg, SparkSession sparkSession, SchemaProvider schemaProvider,
+ public StreamSync(HoodieStreamer.Config cfg, SparkSession sparkSession,
TypedProperties props, HoodieSparkEngineContext hoodieSparkContext, FileSystem fs, Configuration conf,
- Function<SparkRDDWriteClient, Boolean> onInitializingHoodieWriteClient) throws IOException {
+ Function<SparkRDDWriteClient, Boolean> onInitializingHoodieWriteClient, StreamContext streamContext) throws IOException {
this.cfg = cfg;
this.hoodieSparkContext = hoodieSparkContext;
this.sparkSession = sparkSession;
this.fs = fs;
this.onInitializingHoodieWriteClient = onInitializingHoodieWriteClient;
this.props = props;
- this.userProvidedSchemaProvider = schemaProvider;
+ this.userProvidedSchemaProvider = streamContext.getSchemaProvider();
this.processedSchema = new SchemaSet();
this.autoGenerateRecordKeys = KeyGenUtils.enableAutoGenerateRecordKeys(props);
this.keyGenClassName = getKeyGeneratorClassName(new TypedProperties(props));
@@ -285,7 +285,7 @@ public class StreamSync implements Serializable, Closeable {
this.errorWriteFailureStrategy = ErrorTableUtils.getErrorWriteFailureStrategy(props);
}
refreshTimeline();
- Source source = UtilHelpers.createSource(cfg.sourceClassName, props, hoodieSparkContext.jsc(), sparkSession, schemaProvider, metrics);
+ Source source = UtilHelpers.createSource(cfg.sourceClassName, props, hoodieSparkContext.jsc(), sparkSession, metrics, streamContext);
this.formatAdapter = new SourceFormatAdapter(source, this.errorTableWriter, Option.of(props));
Supplier<Option<Schema>> schemaSupplier = schemaProvider == null ? Option::empty : () -> Option.ofNullable(schemaProvider.getSourceSchema());
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/BaseTestKafkaSource.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/BaseTestKafkaSource.java
index b5cbf2738f6..011a1f626b2 100644
--- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/BaseTestKafkaSource.java
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/BaseTestKafkaSource.java
@@ -28,6 +28,8 @@ import org.apache.hudi.utilities.exception.HoodieStreamerException;
import org.apache.hudi.utilities.ingestion.HoodieIngestionMetrics;
import org.apache.hudi.utilities.schema.SchemaProvider;
import org.apache.hudi.utilities.streamer.SourceFormatAdapter;
+import org.apache.hudi.utilities.streamer.SourceProfile;
+import org.apache.hudi.utilities.streamer.SourceProfileSupplier;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -52,6 +54,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
/**
* Generic tests for all {@link KafkaSource} to ensure all implementations properly handle offsets, fetch limits, failure modes, etc.
@@ -60,6 +63,7 @@ abstract class BaseTestKafkaSource extends SparkClientFunctionalTestHarness {
protected static final String TEST_TOPIC_PREFIX = "hoodie_test_";
protected final HoodieIngestionMetrics metrics = mock(HoodieIngestionMetrics.class);
+ protected final Option<SourceProfileSupplier> sourceProfile = Option.of(mock(SourceProfileSupplier.class));
protected SchemaProvider schemaProvider;
protected KafkaTestUtils testUtils;
@@ -277,4 +281,51 @@ abstract class BaseTestKafkaSource extends SparkClientFunctionalTestHarness {
+ " either the data was aged out by Kafka or the topic may have been deleted before all the data in the topic was processed.",
t.getMessage());
}
+
+ @Test
+ public void testKafkaSourceWithOffsetsFromSourceProfile() {
+ // topic setup.
+ final String topic = TEST_TOPIC_PREFIX + "testKafkaSourceWithOffsetRanges";
+ testUtils.createTopic(topic, 2);
+ TypedProperties props = createPropsForKafkaSource(topic, null, "earliest");
+
+ when(sourceProfile.get().getSourceProfile()).thenReturn(new TestSourceProfile(Long.MAX_VALUE, 4, 500));
+ SourceFormatAdapter kafkaSource = createSource(props);
+
+ // Test for empty data.
+ assertEquals(Option.empty(), kafkaSource.fetchNewDataInAvroFormat(Option.empty(), Long.MAX_VALUE).getBatch());
+
+ // Publish messages and assert source has picked up all messages in offsetRanges supplied by input batch profile.
+ sendMessagesToKafka(topic, 1000, 2);
+ InputBatch<JavaRDD<GenericRecord>> fetch1 = kafkaSource.fetchNewDataInAvroFormat(Option.empty(), 900);
+ assertEquals(500, fetch1.getBatch().get().count());
+ }
+
+ static class TestSourceProfile implements SourceProfile<Long> {
+
+ private final long maxSourceBytes;
+ private final int sourcePartitions;
+ private final long numEvents;
+
+ public TestSourceProfile(long maxSourceBytes, int sourcePartitions, long numEvents) {
+ this.maxSourceBytes = maxSourceBytes;
+ this.sourcePartitions = sourcePartitions;
+ this.numEvents = numEvents;
+ }
+
+ @Override
+ public long getMaxSourceBytes() {
+ return maxSourceBytes;
+ }
+
+ @Override
+ public int getSourcePartitions() {
+ return sourcePartitions;
+ }
+
+ @Override
+ public Long getSourceSpecificContext() {
+ return numEvents;
+ }
+ }
}
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestJsonKafkaSource.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestJsonKafkaSource.java
index 60887613d64..166d419001d 100644
--- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestJsonKafkaSource.java
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestJsonKafkaSource.java
@@ -26,11 +26,13 @@ import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.utilities.UtilHelpers;
import org.apache.hudi.utilities.config.HoodieStreamerConfig;
import org.apache.hudi.utilities.config.KafkaSourceConfig;
import org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer;
import org.apache.hudi.utilities.schema.FilebasedSchemaProvider;
import org.apache.hudi.utilities.streamer.BaseErrorTableWriter;
+import org.apache.hudi.utilities.streamer.DefaultStreamContext;
import org.apache.hudi.utilities.streamer.ErrorEvent;
import org.apache.hudi.utilities.streamer.SourceFormatAdapter;
@@ -60,10 +62,10 @@ import scala.Tuple2;
import static org.apache.hudi.config.HoodieErrorTableConfig.ERROR_TABLE_BASE_PATH;
import static org.apache.hudi.config.HoodieErrorTableConfig.ERROR_TARGET_TABLE;
import static org.apache.hudi.utilities.config.KafkaSourceConfig.ENABLE_KAFKA_COMMIT_OFFSET;
+import static org.apache.hudi.utilities.schema.KafkaOffsetPostProcessor.KAFKA_SOURCE_KEY_COLUMN;
import static org.apache.hudi.utilities.schema.KafkaOffsetPostProcessor.KAFKA_SOURCE_OFFSET_COLUMN;
import static org.apache.hudi.utilities.schema.KafkaOffsetPostProcessor.KAFKA_SOURCE_PARTITION_COLUMN;
import static org.apache.hudi.utilities.schema.KafkaOffsetPostProcessor.KAFKA_SOURCE_TIMESTAMP_COLUMN;
-import static org.apache.hudi.utilities.schema.KafkaOffsetPostProcessor.KAFKA_SOURCE_KEY_COLUMN;
import static org.apache.hudi.utilities.testutils.UtilitiesTestBase.Helpers.jsonifyRecords;
import static org.apache.hudi.utilities.testutils.UtilitiesTestBase.Helpers.jsonifyRecordsByPartitions;
import static org.apache.hudi.utilities.testutils.UtilitiesTestBase.Helpers.jsonifyRecordsByPartitionsWithNullKafkaKey;
@@ -104,7 +106,7 @@ public class TestJsonKafkaSource extends BaseTestKafkaSource {
@Override
SourceFormatAdapter createSource(TypedProperties props) {
- return new SourceFormatAdapter(new JsonKafkaSource(props, jsc(), spark(), schemaProvider, metrics));
+ return new SourceFormatAdapter(new JsonKafkaSource(props, jsc(), spark(), metrics, new DefaultStreamContext(schemaProvider, sourceProfile)));
}
// test whether empty messages can be filtered
@@ -356,4 +358,13 @@ public class TestJsonKafkaSource extends BaseTestKafkaSource {
dfWithOffsetInfo.unpersist();
dfWithOffsetInfoAndNullKafkaKey.unpersist();
}
+
+ @Test
+ public void testCreateSource() throws IOException {
+ final String topic = TEST_TOPIC_PREFIX + "testJsonKafkaSourceCreation";
+ testUtils.createTopic(topic, 2);
+ TypedProperties props = createPropsForKafkaSource(topic, null, "earliest");
+ Source jsonKafkaSource = UtilHelpers.createSource(JsonKafkaSource.class.getName(), props, jsc(), spark(), metrics, new DefaultStreamContext(schemaProvider, sourceProfile));
+ assertEquals(Source.SourceType.JSON, jsonKafkaSource.getSourceType());
+ }
}
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestProtoKafkaSource.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestProtoKafkaSource.java
index 52376f89741..b56d87c9263 100644
--- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestProtoKafkaSource.java
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestProtoKafkaSource.java
@@ -25,6 +25,7 @@ import org.apache.hudi.utilities.config.KafkaSourceConfig;
import org.apache.hudi.utilities.config.ProtoClassBasedSchemaProviderConfig;
import org.apache.hudi.utilities.schema.ProtoClassBasedSchemaProvider;
import org.apache.hudi.utilities.schema.SchemaProvider;
+import org.apache.hudi.utilities.streamer.DefaultStreamContext;
import org.apache.hudi.utilities.streamer.SourceFormatAdapter;
import org.apache.hudi.utilities.test.proto.Nested;
import org.apache.hudi.utilities.test.proto.Sample;
@@ -89,7 +90,7 @@ public class TestProtoKafkaSource extends BaseTestKafkaSource {
@Override
SourceFormatAdapter createSource(TypedProperties props) {
this.schemaProvider = new ProtoClassBasedSchemaProvider(props, jsc());
- Source protoKafkaSource = new ProtoKafkaSource(props, jsc(), spark(), schemaProvider, metrics);
+ Source protoKafkaSource = new ProtoKafkaSource(props, jsc(), spark(), metrics, new DefaultStreamContext(schemaProvider, sourceProfile));
return new SourceFormatAdapter(protoKafkaSource);
}
(hudi) 18/18: [HUDI-7441] Move `getWritePartitionPaths` method to common module to decouple hive dependency (#10744)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 855960f819e544d2f743adcf22dc0e0b1f1a0fb6
Author: stayrascal <st...@users.noreply.github.com>
AuthorDate: Tue Feb 27 13:44:43 2024 +0800
[HUDI-7441] Move `getWritePartitionPaths` method to common module to decouple hive dependency (#10744)
Co-authored-by: wuzhiping <wu...@bytedance.com>
---
.../org/apache/hudi/metadata/HoodieTableMetadataUtil.java | 13 +++++++++++++
.../java/org/apache/hudi/source/IncrementalInputSplits.java | 4 ++--
.../hadoop/realtime/HoodieMergeOnReadTableInputFormat.java | 3 ++-
.../apache/hudi/hadoop/utils/HoodieInputFormatUtils.java | 13 -------------
.../org/apache/hudi/MergeOnReadIncrementalRelation.scala | 3 ++-
5 files changed, 19 insertions(+), 17 deletions(-)
diff --git a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
index 0aa11042ab9..d364ce77054 100644
--- a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
+++ b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
@@ -425,6 +425,19 @@ public class HoodieTableMetadataUtil {
.collect(Collectors.toList());
}
+ /**
+ * Returns all the incremental write partition paths as a set with the given commits metadata.
+ *
+ * @param metadataList The commits metadata
+ * @return the partition path set
+ */
+ public static Set<String> getWritePartitionPaths(List<HoodieCommitMetadata> metadataList) {
+ return metadataList.stream()
+ .map(HoodieCommitMetadata::getWritePartitionPaths)
+ .flatMap(Collection::stream)
+ .collect(Collectors.toSet());
+ }
+
/**
* Convert commit action metadata to bloom filter records.
*
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java
index 05d11bf746f..e179e532078 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java
@@ -35,7 +35,7 @@ import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.configuration.OptionsResolver;
-import org.apache.hudi.hadoop.utils.HoodieInputFormatUtils;
+import org.apache.hudi.metadata.HoodieTableMetadataUtil;
import org.apache.hudi.sink.partitioner.profile.WriteProfiles;
import org.apache.hudi.source.prune.PartitionPruners;
import org.apache.hudi.table.format.cdc.CdcInputSplit;
@@ -500,7 +500,7 @@ public class IncrementalInputSplits implements Serializable {
* @return the set of read partitions
*/
private Set<String> getReadPartitions(List<HoodieCommitMetadata> metadataList) {
- Set<String> partitions = HoodieInputFormatUtils.getWritePartitionPaths(metadataList);
+ Set<String> partitions = HoodieTableMetadataUtil.getWritePartitionPaths(metadataList);
// apply partition push down
if (this.partitionPruner != null) {
Set<String> selectedPartitions = this.partitionPruner.filter(partitions);
diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/HoodieMergeOnReadTableInputFormat.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/HoodieMergeOnReadTableInputFormat.java
index 3719718e95a..e367cefd7fc 100644
--- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/HoodieMergeOnReadTableInputFormat.java
+++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/HoodieMergeOnReadTableInputFormat.java
@@ -54,6 +54,7 @@ import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.SplitLocationInfo;
import org.apache.hadoop.mapreduce.Job;
+import org.apache.hudi.metadata.HoodieTableMetadataUtil;
import java.io.IOException;
import java.util.ArrayList;
@@ -195,7 +196,7 @@ public class HoodieMergeOnReadTableInputFormat extends HoodieCopyOnWriteTableInp
// build fileGroup from fsView
Path basePath = new Path(tableMetaClient.getBasePath());
// filter affectedPartition by inputPaths
- List<String> affectedPartition = HoodieInputFormatUtils.getWritePartitionPaths(metadataList).stream()
+ List<String> affectedPartition = HoodieTableMetadataUtil.getWritePartitionPaths(metadataList).stream()
.filter(k -> k.isEmpty() ? inputPaths.contains(basePath) : inputPaths.contains(new Path(basePath, k))).collect(Collectors.toList());
if (affectedPartition.isEmpty()) {
return result;
diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
index 8922b837871..4ab72701a11 100644
--- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
+++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
@@ -514,19 +514,6 @@ public class HoodieInputFormatUtils {
return fullPathToFileStatus.values().toArray(new FileStatus[0]);
}
- /**
- * Returns all the incremental write partition paths as a set with the given commits metadata.
- *
- * @param metadataList The commits metadata
- * @return the partition path set
- */
- public static Set<String> getWritePartitionPaths(List<HoodieCommitMetadata> metadataList) {
- return metadataList.stream()
- .map(HoodieCommitMetadata::getWritePartitionPaths)
- .flatMap(Collection::stream)
- .collect(Collectors.toSet());
- }
-
public static HoodieRealtimeFileSplit createRealtimeFileSplit(HoodieRealtimePath path, long start, long length, String[] hosts) {
try {
return new HoodieRealtimeFileSplit(new FileSplit(path, start, length, hosts), path);
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala
index 2904992fdef..93d279baab1 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala
@@ -26,9 +26,10 @@ import org.apache.hudi.common.table.timeline.TimelineUtils.HollowCommitHandling.
import org.apache.hudi.common.table.timeline.TimelineUtils.{HollowCommitHandling, getCommitMetadata, handleHollowCommitIfNeeded}
import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
import org.apache.hudi.common.table.view.HoodieTableFileSystemView
+import org.apache.hudi.metadata.HoodieTableMetadataUtil.getWritePartitionPaths
import org.apache.hudi.common.util.StringUtils
import org.apache.hudi.exception.HoodieException
-import org.apache.hudi.hadoop.utils.HoodieInputFormatUtils.{getWritePartitionPaths, listAffectedFilesForCommits}
+import org.apache.hudi.hadoop.utils.HoodieInputFormatUtils.listAffectedFilesForCommits
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.catalyst.InternalRow
(hudi) 03/18: [HUDI-7275] Separate use of HoodieTimelineTimeZone.UTC and LOCAL in tests to prevent infinite loops (#10738)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit d0b34f0857684ce060e84ab2421ec14882aeae5f
Author: Geser Dugarov, PhD <ge...@gmail.com>
AuthorDate: Sat Feb 24 11:34:54 2024 +0700
[HUDI-7275] Separate use of HoodieTimelineTimeZone.UTC and LOCAL in tests to prevent infinite loops (#10738)
---
.../org/apache/hudi/TestHoodieSparkSqlWriter.scala | 59 ++-------------
.../apache/hudi/TestHoodieSparkSqlWriterUtc.scala | 85 ++++++++++++++++++++++
2 files changed, 91 insertions(+), 53 deletions(-)
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriter.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriter.scala
index 1c6766063d2..c57785e5ffe 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriter.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriter.scala
@@ -23,7 +23,6 @@ import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.HoodieSparkUtils.gteqSpark3_0
import org.apache.hudi.client.SparkRDDWriteClient
import org.apache.hudi.common.model._
-import org.apache.hudi.common.table.timeline.HoodieInstantTimeGenerator
import org.apache.hudi.common.table.{HoodieTableConfig, HoodieTableMetaClient, TableSchemaResolver}
import org.apache.hudi.common.testutils.HoodieTestDataGenerator
import org.apache.hudi.config.{HoodieBootstrapConfig, HoodieIndexConfig, HoodieWriteConfig}
@@ -40,7 +39,7 @@ import org.apache.spark.sql.functions.{expr, lit}
import org.apache.spark.sql.hudi.HoodieSparkSessionExtension
import org.apache.spark.sql.hudi.command.SqlKeyGenerator
import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertNotNull, assertNull, assertTrue, fail}
-import org.junit.jupiter.api.{AfterEach, BeforeEach, Disabled, Test}
+import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.Arguments.arguments
import org.junit.jupiter.params.provider._
@@ -50,15 +49,16 @@ import org.scalatest.Assertions.assertThrows
import org.scalatest.Matchers.{be, convertToAnyShouldWrapper, intercept}
import java.io.IOException
-import java.time.format.DateTimeFormatterBuilder
-import java.time.temporal.ChronoField
-import java.time.{Instant, ZoneId}
-import java.util.{Collections, Date, TimeZone, UUID}
+import java.time.Instant
+import java.util.{Collections, Date, UUID}
import scala.collection.JavaConversions._
import scala.collection.JavaConverters
/**
* Test suite for SparkSqlWriter class.
+ * All cases of using of {@link HoodieTimelineTimeZone.UTC} should be done in a separate test class {@link TestHoodieSparkSqlWriterUtc}.
+ * Otherwise UTC tests will generate infinite loops, if there is any initiated test with time zone that is greater then UTC+0.
+ * The reason is in a saved value in the heap of static {@link org.apache.hudi.common.table.timeline.HoodieInstantTimeGenerator.lastInstantTime}.
*/
class TestHoodieSparkSqlWriter {
var spark: SparkSession = _
@@ -1336,53 +1336,6 @@ def testBulkInsertForDropPartitionColumn(): Unit = {
assert(exc.getMessage.contains("Consistent hashing bucket index does not work with COW table. Use simple bucket index or an MOR table."))
}
- /*
- * Test case for instant is generated with commit timezone when TIMELINE_TIMEZONE set to UTC
- * related to HUDI-5978
- * Issue [HUDI-7275] is tracking this test being disabled
- */
- @Disabled
- def testInsertDatasetWithTimelineTimezoneUTC(): Unit = {
- val defaultTimezone = TimeZone.getDefault
- try {
- val fooTableModifier = commonTableModifier.updated(DataSourceWriteOptions.OPERATION.key, DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL)
- .updated(DataSourceWriteOptions.INSERT_DROP_DUPS.key, "false")
- .updated(HoodieTableConfig.TIMELINE_TIMEZONE.key, "UTC") // utc timezone
-
- // generate the inserts
- val schema = DataSourceTestUtils.getStructTypeExampleSchema
- val structType = AvroConversionUtils.convertAvroSchemaToStructType(schema)
- val records = DataSourceTestUtils.generateRandomRows(100)
- val recordsSeq = convertRowListToSeq(records)
- val df = spark.createDataFrame(sc.parallelize(recordsSeq), structType)
-
- // get UTC instant before write
- val beforeWriteInstant = Instant.now()
-
- // set local timezone to America/Los_Angeles(UTC-7)
- TimeZone.setDefault(TimeZone.getTimeZone("America/Los_Angeles"))
-
- // write to Hudi
- val (success, writeInstantTimeOpt, _, _, _, hoodieTableConfig) = HoodieSparkSqlWriter.write(sqlContext, SaveMode.Append, fooTableModifier, df)
- assertTrue(success)
- val hoodieTableTimelineTimezone = HoodieTimelineTimeZone.valueOf(hoodieTableConfig.getString(HoodieTableConfig.TIMELINE_TIMEZONE))
- assertEquals(hoodieTableTimelineTimezone, HoodieTimelineTimeZone.UTC)
-
- val utcFormatter = new DateTimeFormatterBuilder()
- .appendPattern(HoodieInstantTimeGenerator.SECS_INSTANT_TIMESTAMP_FORMAT)
- .appendValue(ChronoField.MILLI_OF_SECOND, 3)
- .toFormatter
- .withZone(ZoneId.of("UTC"))
- // instant parsed by UTC timezone
- val writeInstant = Instant.from(utcFormatter.parse(writeInstantTimeOpt.get()))
-
- assertTrue(beforeWriteInstant.toEpochMilli < writeInstant.toEpochMilli,
- s"writeInstant(${writeInstant.toEpochMilli}) must always be greater than beforeWriteInstant(${beforeWriteInstant.toEpochMilli}) if writeInstant was generated with UTC timezone")
- } finally {
- TimeZone.setDefault(defaultTimezone)
- }
- }
-
private def fetchActualSchema(): Schema = {
val tableMetaClient = HoodieTableMetaClient.builder()
.setConf(spark.sparkContext.hadoopConfiguration)
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriterUtc.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriterUtc.scala
new file mode 100644
index 00000000000..df8614f5e2a
--- /dev/null
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriterUtc.scala
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi
+
+import org.apache.hudi.common.model.HoodieTimelineTimeZone
+import org.apache.hudi.common.table.HoodieTableConfig
+import org.apache.hudi.common.table.timeline.HoodieInstantTimeGenerator
+import org.apache.hudi.testutils.DataSourceTestUtils
+import org.apache.spark.sql.SaveMode
+import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
+import org.junit.jupiter.api.Test
+
+import java.time.{Instant, ZoneId}
+import java.time.format.DateTimeFormatterBuilder
+import java.time.temporal.ChronoField
+import java.util.TimeZone
+
+/**
+ * Test suite for SparkSqlWriter class for all cases of using of {@link HoodieTimelineTimeZone.UTC}.
+ * Using of {@link HoodieTimelineTimeZone.LOCAL} here could lead to infinite loops, because it could save
+ * value of static {@link HoodieInstantTimeGenerator.lastInstantTime} in the heap,
+ * which will be greater than instant time for {@link HoodieTimelineTimeZone.UTC}.
+ */
+class TestHoodieSparkSqlWriterUtc extends TestHoodieSparkSqlWriter {
+ /*
+ * Test case for instant is generated with commit timezone when TIMELINE_TIMEZONE set to UTC
+ * related to HUDI-5978
+ */
+ @Test
+ def testInsertDatasetWithTimelineTimezoneUTC(): Unit = {
+ val defaultTimezone = TimeZone.getDefault
+ try {
+ val fooTableModifier = commonTableModifier.updated(DataSourceWriteOptions.OPERATION.key, DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL)
+ .updated(DataSourceWriteOptions.INSERT_DROP_DUPS.key, "false")
+ .updated(HoodieTableConfig.TIMELINE_TIMEZONE.key, "UTC") // utc timezone
+
+ // generate the inserts
+ val schema = DataSourceTestUtils.getStructTypeExampleSchema
+ val structType = AvroConversionUtils.convertAvroSchemaToStructType(schema)
+ val records = DataSourceTestUtils.generateRandomRows(100)
+ val recordsSeq = convertRowListToSeq(records)
+ val df = spark.createDataFrame(sc.parallelize(recordsSeq), structType)
+
+ // get UTC instant before write
+ val beforeWriteInstant = Instant.now()
+
+ // set local timezone to America/Los_Angeles(UTC-7)
+ TimeZone.setDefault(TimeZone.getTimeZone("Asia/Novosibirsk"))
+
+ // write to Hudi
+ val (success, writeInstantTimeOpt, _, _, _, hoodieTableConfig) = HoodieSparkSqlWriter.write(sqlContext, SaveMode.Append, fooTableModifier, df)
+ assertTrue(success)
+ val hoodieTableTimelineTimezone = HoodieTimelineTimeZone.valueOf(hoodieTableConfig.getString(HoodieTableConfig.TIMELINE_TIMEZONE))
+ assertEquals(hoodieTableTimelineTimezone, HoodieTimelineTimeZone.UTC)
+
+ val utcFormatter = new DateTimeFormatterBuilder()
+ .appendPattern(HoodieInstantTimeGenerator.SECS_INSTANT_TIMESTAMP_FORMAT)
+ .appendValue(ChronoField.MILLI_OF_SECOND, 3)
+ .toFormatter
+ .withZone(ZoneId.of("UTC"))
+ // instant parsed by UTC timezone
+ val writeInstant = Instant.from(utcFormatter.parse(writeInstantTimeOpt.get()))
+
+ assertTrue(beforeWriteInstant.toEpochMilli < writeInstant.toEpochMilli,
+ s"writeInstant(${writeInstant.toEpochMilli}) must always be greater than beforeWriteInstant(${beforeWriteInstant.toEpochMilli}) if writeInstant was generated with UTC timezone")
+ } finally {
+ TimeZone.setDefault(defaultTimezone)
+ }
+ }
+}