yujunjun/OpenMetadata
1
0
Fork 0
mirror of https://github.com/open-metadata/OpenMetadata.git synced 2025-08-31 12:39:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

Search Index Issues (#18648)

Browse Source 
* Search Index Issues

* Log Thread

* Make logs debug

* Remove Current Therad Name
This commit is contained in:
Mohit Yadav 2024-11-14 22:19:35 +05:30 committed by GitHub
parent 75d417d267
commit 560e28df76
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 46 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -100,6 +100,7 @@ import org.quartz.JobExecutionContext;
public class SearchIndexApp extends AbstractNativeApplication {
private static final String ALL = "all";
public static final Set<String> ALL_ENTITIES =
Set.of(
TABLE,
@ -162,8 +163,9 @@ public class SearchIndexApp extends AbstractNativeApplication {
private volatile boolean stopped = false;
private ExecutorService consumerExecutor;
private ExecutorService producerExecutor;
private ExecutorService jobExecutor = Executors.newFixedThreadPool(2);
private final ExecutorService jobExecutor = Executors.newCachedThreadPool();
private BlockingQueue<IndexingTask<?>> taskQueue = new LinkedBlockingQueue<>(100);
private BlockingQueue<Runnable> producerQueue = new LinkedBlockingQueue<>(100);
private final AtomicReference<Stats> searchIndexStats = new AtomicReference<>();
private final AtomicReference<Integer> batchSize = new AtomicReference<>(5);
@ -289,6 +291,7 @@ public class SearchIndexApp extends AbstractNativeApplication {
LOG.info("Starting reindexing with {} producers and {} consumers.", numProducers, numConsumers);
taskQueue = new LinkedBlockingQueue<>(jobData.getQueueSize());
producerQueue = new LinkedBlockingQueue<>(jobData.getQueueSize());
consumerExecutor =
new ThreadPoolExecutor(
numConsumers,
@ -302,7 +305,7 @@ public class SearchIndexApp extends AbstractNativeApplication {
numProducers,
0L,
TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<>(jobData.getQueueSize()),
producerQueue,
new ThreadPoolExecutor.CallerRunsPolicy());
try {
@ -321,8 +324,8 @@ public class SearchIndexApp extends AbstractNativeApplication {
throws InterruptedException {
int numConsumers = jobData.getConsumerThreads();
CountDownLatch producerLatch = new CountDownLatch(getTotalLatchCount(jobData.getEntities()));
jobExecutor.submit(() -> submitProducerTask(producerLatch));
jobExecutor.submit(() -> submitConsumerTask(jobExecutionContext));
submitProducerTask(producerLatch);
submitConsumerTask(jobExecutionContext);
producerLatch.await();
sendPoisonPills(numConsumers);
@ -330,6 +333,8 @@ public class SearchIndexApp extends AbstractNativeApplication {
private void submitProducerTask(CountDownLatch producerLatch) {
for (String entityType : jobData.getEntities()) {
jobExecutor.submit(
() -> {
try {
reCreateIndexes(entityType);
int totalEntityRecords = getTotalEntityRecords(entityType);
@ -337,14 +342,21 @@ public class SearchIndexApp extends AbstractNativeApplication {
int noOfThreads = calculateNumberOfThreads(totalEntityRecords);
if (totalEntityRecords > 0) {
for (int i = 0; i < noOfThreads; i++) {
LOG.debug(
"Submitting producer task current queue size: {}", producerQueue.size());
int currentOffset = i * batchSize.get();
producerExecutor.submit(
() -> {
try {
LOG.debug(
"Running Task for CurrentOffset: {}, Producer Latch Down, Current : {}",
currentOffset,
producerLatch.getCount());
processReadTask(entityType, source, currentOffset);
} catch (Exception e) {
LOG.error("Error processing entity type {}", entityType, e);
} finally {
LOG.debug("Producer Latch Down, Current : {}", producerLatch.getCount());
producerLatch.countDown();
}
});
@ -353,6 +365,7 @@ public class SearchIndexApp extends AbstractNativeApplication {
} catch (Exception e) {
LOG.error("Error processing entity type {}", entityType, e);
}
});
}
}
@ -572,6 +585,7 @@ public class SearchIndexApp extends AbstractNativeApplication {
public void stopJob() {
LOG.info("Stopping reindexing job.");
stopped = true;
shutdownExecutor(jobExecutor, "JobExecutor", 60, TimeUnit.SECONDS);
shutdownExecutor(producerExecutor, "ProducerExecutor", 60, TimeUnit.SECONDS);
shutdownExecutor(consumerExecutor, "ConsumerExecutor", 60, TimeUnit.SECONDS);
}
@ -679,12 +693,12 @@ public class SearchIndexApp extends AbstractNativeApplication {
private void processReadTask(String entityType, Source<?> source, int offset) {
try {
Object resultList = source.readWithCursor(RestUtil.encodeCursor(String.valueOf(offset)));
LOG.info("Read Entities with CurrentOffset: {}", offset);
if (resultList != null) {
ResultList<?> entities = extractEntities(entityType, resultList);
if (!nullOrEmpty(entities.getData())) {
LOG.info(
"Creating Indexing Task for entityType: {}, current offset: {}", entityType, offset);
createIndexingTask(entityType, entities, offset);
IndexingTask<?> task = new IndexingTask<>(entityType, entities, offset);
taskQueue.put(task);
}
}
} catch (InterruptedException e) {
@ -708,13 +722,7 @@ public class SearchIndexApp extends AbstractNativeApplication {
}
}
private void createIndexingTask(String entityType, ResultList<?> entities, int offset)
throws InterruptedException {
IndexingTask<?> task = new IndexingTask<>(entityType, entities, offset);
taskQueue.put(task);
}
private synchronized int calculateNumberOfThreads(int totalEntityRecords) {
private int calculateNumberOfThreads(int totalEntityRecords) {
int mod = totalEntityRecords % batchSize.get();
if (mod == 0) {
return totalEntityRecords / batchSize.get();
@ -732,7 +740,7 @@ public class SearchIndexApp extends AbstractNativeApplication {
}
}
private synchronized int getRemainingRecordsToProcess(String entityType) {
private int getRemainingRecordsToProcess(String entityType) {
StepStats entityStats =
((StepStats)
searchIndexStats.get().getEntityStats().getAdditionalProperties().get(entityType));