yujunjun/OpenMetadata
1
0
Fork 0
mirror of https://github.com/open-metadata/OpenMetadata.git synced 2025-08-30 20:06:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add Opensearch max payload Size (#17392)

Browse Source
This commit is contained in:
Mohit Yadav 2024-08-12 11:34:52 +05:30 committed by GitHub
parent 05baba9dc1
commit df9753c7bb
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 14 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/insights/workflows/dataAssets/DataAssetsWorkflow.java
View File

@ -122,10 +122,18 @@ public class DataAssetsWorkflow {
this.entityEnricher = new DataInsightsEntityEnricherProcessor(totalRecords); this.entityEnricher = new DataInsightsEntityEnricherProcessor(totalRecords);
if (searchRepository.getSearchType().equals(ElasticSearchConfiguration.SearchType.OPENSEARCH)) { if (searchRepository.getSearchType().equals(ElasticSearchConfiguration.SearchType.OPENSEARCH)) {
this.entityProcessor = new DataInsightsOpenSearchProcessor(totalRecords); this.entityProcessor = new DataInsightsOpenSearchProcessor(totalRecords);
this.searchIndexSink = new OpenSearchIndexSink(searchRepository, totalRecords); this.searchIndexSink =
new OpenSearchIndexSink(
searchRepository,
totalRecords,
searchRepository.getElasticSearchConfiguration().getPayLoadSize());
} else { } else {
this.entityProcessor = new DataInsightsElasticSearchProcessor(totalRecords); this.entityProcessor = new DataInsightsElasticSearchProcessor(totalRecords);
this.searchIndexSink = new ElasticSearchIndexSink(searchRepository, totalRecords); this.searchIndexSink =
new ElasticSearchIndexSink(
searchRepository,
totalRecords,
searchRepository.getElasticSearchConfiguration().getPayLoadSize());
} }
} }

3
openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/searchIndex/SearchIndexApp.java
View File

@ -195,7 +195,8 @@ public class SearchIndexApp extends AbstractNativeApplication {
if (searchRepository.getSearchType().equals(ElasticSearchConfiguration.SearchType.OPENSEARCH)) { if (searchRepository.getSearchType().equals(ElasticSearchConfiguration.SearchType.OPENSEARCH)) {
this.entityProcessor = new OpenSearchEntitiesProcessor(totalRecords); this.entityProcessor = new OpenSearchEntitiesProcessor(totalRecords);
this.entityTimeSeriesProcessor = new OpenSearchEntityTimeSeriesProcessor(totalRecords); this.entityTimeSeriesProcessor = new OpenSearchEntityTimeSeriesProcessor(totalRecords);
this.searchIndexSink = new OpenSearchIndexSink(searchRepository, totalRecords); this.searchIndexSink =
new OpenSearchIndexSink(searchRepository, totalRecords, jobData.getPayLoadSize());
} else { } else {
this.entityProcessor = new ElasticSearchEntitiesProcessor(totalRecords); this.entityProcessor = new ElasticSearchEntitiesProcessor(totalRecords);
this.entityTimeSeriesProcessor = new ElasticSearchEntityTimeSeriesProcessor(totalRecords); this.entityTimeSeriesProcessor = new ElasticSearchEntityTimeSeriesProcessor(totalRecords);

6
openmetadata-service/src/main/java/org/openmetadata/service/search/elasticsearch/ElasticSearchIndexSink.java
View File

@ -30,12 +30,6 @@ public class ElasticSearchIndexSink implements Sink<BulkRequest, BulkResponse> {
private final SearchRepository searchRepository; private final SearchRepository searchRepository;
private final int maxPayLoadSizeInBytes; private final int maxPayLoadSizeInBytes;
public ElasticSearchIndexSink(SearchRepository searchRepository, int total) {
this.searchRepository = searchRepository;
this.maxPayLoadSizeInBytes = searchRepository.getElasticSearchConfiguration().getPayLoadSize();
this.stats.withTotalRecords(total).withSuccessRecords(0).withFailedRecords(0);
}
public ElasticSearchIndexSink( public ElasticSearchIndexSink(
SearchRepository searchRepository, int total, int maxPayLoadSizeInBytes) { SearchRepository searchRepository, int total, int maxPayLoadSizeInBytes) {
this.searchRepository = searchRepository; this.searchRepository = searchRepository;

4
openmetadata-service/src/main/java/org/openmetadata/service/search/opensearch/OpenSearchIndexSink.java
View File

@ -31,9 +31,9 @@ public class OpenSearchIndexSink implements Sink<BulkRequest, BulkResponse> {
private final int maxPayLoadSizeInBytes; private final int maxPayLoadSizeInBytes;
public OpenSearchIndexSink(SearchRepository repository, int total) { public OpenSearchIndexSink(SearchRepository repository, int total, int maxPayLoadSizeInBytes) {
this.searchRepository = repository; this.searchRepository = repository;
this.maxPayLoadSizeInBytes = searchRepository.getElasticSearchConfiguration().getPayLoadSize(); this.maxPayLoadSizeInBytes = maxPayLoadSizeInBytes;
this.stats.withTotalRecords(total).withSuccessRecords(0).withFailedRecords(0); this.stats.withTotalRecords(total).withSuccessRecords(0).withFailedRecords(0);
} }