Skip to content

Commit d3101dc

Browse files
Chandra Kanth PeravelliChandra Kanth Peravelli
Chandra Kanth Peravelli
authored and
Chandra Kanth Peravelli
committed
ATLAS-5093: Ensure executors are avialable and alive before submitting import tasks
1 parent 2eafacd commit d3101dc

4 files changed

Lines changed: 195 additions & 14 deletions

File tree

‎repository/src/main/java/org/apache/atlas/repository/patches/UpdateCompositeIndexStatusPatch.java‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@
1919

2020
import org.apache.atlas.AtlasConfiguration;
2121
import org.apache.atlas.exception.AtlasBaseException;
22-
import org.apache.atlas.repository.graphdb.AtlasGraph;
2322
import org.apache.atlas.repository.graphdb.AtlasGraphManagement;
2423
import org.slf4j.Logger;
2524
import org.slf4j.LoggerFactory;

‎repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEnumDefStoreV2.java‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -313,7 +313,6 @@ private void createPropertyKeys(AtlasEnumDef enumDef) throws AtlasBaseException
313313
Exception err = null;
314314

315315
try {
316-
317316
// create property keys first
318317
for (AtlasEnumElementDef element : enumDef.getElementDefs()) {
319318
// Validate the enum element

‎webapp/src/main/java/org/apache/atlas/notification/ImportTaskListenerImpl.java‎

Lines changed: 39 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -58,13 +58,14 @@
5858
@Order(8)
5959
@DependsOn(value = "notificationHookConsumer")
6060
public class ImportTaskListenerImpl implements Service, ActiveStateChangeHandler, ImportTaskListener {
61-
private static final Logger LOG = LoggerFactory.getLogger(ImportTaskListenerImpl.class);
61+
private static final Logger LOG = LoggerFactory.getLogger(ImportTaskListenerImpl.class);
6262

63-
private static final String THREADNAME_PREFIX = ImportTaskListener.class.getSimpleName();
64-
private static final int ASYNC_IMPORT_PERMITS = 1; // Only one asynchronous import task is permitted
63+
private static final String THREADNAME_PREFIX = ImportTaskListener.class.getSimpleName();
64+
private static final int ASYNC_IMPORT_PERMITS = 1; // Only one asynchronous import task is permitted
6565

66+
private volatile boolean isActiveInstance = true;
67+
private volatile ExecutorService executorService; // Single-thread executor for sequential processing
6668
private final BlockingQueue<String> requestQueue; // Blocking queue for requests
67-
private final ExecutorService executorService; // Single-thread executor for sequential processing
6869
private final AsyncImportService asyncImportService;
6970
private final NotificationHookConsumer notificationHookConsumer;
7071
private final Semaphore asyncImportSemaphore;
@@ -81,8 +82,6 @@ public ImportTaskListenerImpl(AsyncImportService asyncImportService, Notificatio
8182
this.requestQueue = requestQueue;
8283
this.asyncImportSemaphore = new Semaphore(ASYNC_IMPORT_PERMITS);
8384
this.applicationProperties = ApplicationProperties.get();
84-
this.executorService = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder().setNameFormat(THREADNAME_PREFIX + " thread-%d")
85-
.setUncaughtExceptionHandler((thread, throwable) -> LOG.error("Uncaught exception in thread {}: {}", thread.getName(), throwable.getMessage(), throwable)).build());
8685
}
8786

8887
@Override
@@ -109,11 +108,13 @@ public void stop() throws AtlasException {
109108
public void instanceIsActive() {
110109
LOG.info("Reacting to active state: initializing Kafka consumers");
111110

111+
isActiveInstance = true;
112112
startInternal();
113113
}
114114

115115
@Override
116116
public void instanceIsPassive() {
117+
isActiveInstance = false;
117118
try {
118119
stopImport();
119120
} finally {
@@ -166,6 +167,10 @@ public void onCompleteImportRequest(String importId) {
166167
public void stopImport() {
167168
LOG.info("Shutting down import processor...");
168169

170+
if (executorService == null) {
171+
LOG.info("Executor service is already null, nothing to shut down.");
172+
return;
173+
}
169174
executorService.shutdown(); // Initiate an orderly shutdown
170175

171176
try {
@@ -217,6 +222,10 @@ void startNextImportInQueue() {
217222
void startAsyncImportIfAvailable(String importId) {
218223
LOG.info("==> startAsyncImportIfAvailable()");
219224

225+
if (!isActiveInstance) {
226+
LOG.warn("Import processing attempted while instance is passive. Skipping import.");
227+
return;
228+
}
220229
try {
221230
if (!asyncImportSemaphore.tryAcquire()) {
222231
LOG.info("An async import is in progress, import request is queued");
@@ -232,7 +241,12 @@ void startAsyncImportIfAvailable(String importId) {
232241
return;
233242
}
234243

235-
executorService.submit(() -> startImportConsumer(nextImport));
244+
ExecutorService exec = ensureExecutorAlive();
245+
if (exec != null) {
246+
exec.submit(() -> startImportConsumer(nextImport));
247+
} else {
248+
LOG.warn("No executor available to process import task (instance is passive).");
249+
}
236250
} catch (Exception e) {
237251
LOG.error("Error while starting the next import, releasing the lock if held", e);
238252

@@ -296,6 +310,24 @@ boolean isNotValidImportRequest(AtlasAsyncImportRequest importRequest) {
296310
(!ImportStatus.WAITING.equals(importRequest.getStatus()) && !ImportStatus.PROCESSING.equals(importRequest.getStatus()));
297311
}
298312

313+
@VisibleForTesting
314+
ExecutorService ensureExecutorAlive() {
315+
if (!isActiveInstance) {
316+
LOG.warn("Attempted to create executor while instance is passive. No executor will be created.");
317+
return null;
318+
}
319+
if (executorService == null || executorService.isShutdown() || executorService.isTerminated()) {
320+
synchronized (this) {
321+
if (executorService == null || executorService.isShutdown() || executorService.isTerminated()) {
322+
executorService = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder().setNameFormat(THREADNAME_PREFIX + " thread-%d")
323+
.setUncaughtExceptionHandler((thread, throwable) -> LOG.error("Uncaught exception in thread {}: {}", thread.getName(), throwable.getMessage(), throwable)).build());
324+
LOG.info("ExecutorService was recreated.");
325+
}
326+
}
327+
}
328+
return executorService;
329+
}
330+
299331
void populateRequestQueue() {
300332
LOG.info("==> populateRequestQueue()");
301333

‎webapp/src/test/java/org/apache/atlas/notification/ImportTaskListenerImplTest.java‎

Lines changed: 156 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,9 @@
3737
import java.util.List;
3838
import java.util.concurrent.BlockingDeque;
3939
import java.util.concurrent.CountDownLatch;
40+
import java.util.concurrent.CyclicBarrier;
4041
import java.util.concurrent.ExecutorService;
42+
import java.util.concurrent.Future;
4143
import java.util.concurrent.Semaphore;
4244
import java.util.concurrent.TimeUnit;
4345
import java.util.concurrent.atomic.AtomicBoolean;
@@ -59,11 +61,7 @@
5961
import static org.mockito.Mockito.times;
6062
import static org.mockito.Mockito.verify;
6163
import static org.mockito.Mockito.when;
62-
import static org.testng.Assert.assertEquals;
63-
import static org.testng.Assert.assertNotNull;
64-
import static org.testng.Assert.assertNull;
65-
import static org.testng.Assert.assertTrue;
66-
import static org.testng.Assert.fail;
64+
import static org.testng.Assert.*;
6765

6866
public class ImportTaskListenerImplTest {
6967
private static final String VALID_IMPORT_ID = "valid-id";
@@ -286,6 +284,7 @@ public void testStartAsyncImportIfAvailable_WithInvalidStatus() throws Exception
286284

287285
@Test
288286
public void testStartImportConsumer_Successful() throws Exception {
287+
Mockito.doReturn("import123").when(importRequest).getImportId();
289288
when(importRequest.getStatus()).thenReturn(WAITING);
290289
when(importRequest.getTopicName()).thenReturn("topic1");
291290

@@ -565,6 +564,158 @@ public void testStartInternalIsNonBlocking() throws InterruptedException {
565564
blockStartNextLatch.countDown();
566565
}
567566

567+
@Test
568+
public void testImportNotProcessedWhenPassive() throws Exception {
569+
Mockito.doReturn("import123").when(importRequest).getImportId();
570+
when(importRequest.getStatus()).thenReturn(WAITING);
571+
when(requestQueue.poll(anyLong(), any(TimeUnit.class))).thenReturn("import123");
572+
importTaskListener.instanceIsPassive();
573+
importTaskListener.onReceiveImportRequest(importRequest);
574+
Thread.sleep(200);
575+
verify(notificationHookConsumer, never()).startAsyncImportConsumer(any(), anyString(), anyString());
576+
}
577+
578+
@Test
579+
public void testExecutorNotRecreatedWhenPassive() throws Exception {
580+
Mockito.doReturn("import123").when(importRequest).getImportId();
581+
when(importRequest.getStatus()).thenReturn(WAITING);
582+
when(requestQueue.poll(anyLong(), any(TimeUnit.class))).thenReturn("import123");
583+
when(importRequest.getStatus()).thenReturn(WAITING);
584+
when(requestQueue.poll(anyLong(), any(TimeUnit.class))).thenReturn("import123");
585+
importTaskListener.instanceIsPassive();
586+
Field executorField = ImportTaskListenerImpl.class.getDeclaredField("executorService");
587+
executorField.setAccessible(true);
588+
ExecutorService exec = (ExecutorService) executorField.get(importTaskListener);
589+
if (exec != null) {
590+
exec.shutdownNow();
591+
}
592+
importTaskListener.onReceiveImportRequest(importRequest);
593+
Thread.sleep(200);
594+
ExecutorService execAfter = (ExecutorService) executorField.get(importTaskListener);
595+
// Should remain null when passive
596+
assertTrue(execAfter == null);
597+
}
598+
599+
@Test
600+
public void testExecutorRecreatedWhenActive() throws Exception {
601+
when(importRequest.getStatus()).thenReturn(WAITING);
602+
when(requestQueue.poll(anyLong(), any(TimeUnit.class))).thenReturn("import123");
603+
importTaskListener.instanceIsActive();
604+
Field executorField = ImportTaskListenerImpl.class.getDeclaredField("executorService");
605+
executorField.setAccessible(true);
606+
ExecutorService exec = (ExecutorService) executorField.get(importTaskListener);
607+
if (exec != null) {
608+
exec.shutdownNow();
609+
}
610+
importTaskListener.onReceiveImportRequest(importRequest);
611+
Thread.sleep(200);
612+
ExecutorService execAfter = (ExecutorService) executorField.get(importTaskListener);
613+
assertNotNull(execAfter);
614+
assertTrue(!execAfter.isShutdown() && !execAfter.isTerminated());
615+
}
616+
617+
@Test
618+
public void ensureExecutorAliveCreatesSingleInstanceUnderConcurrency() throws Exception {
619+
// Ensure active mode and a clean executor state
620+
importTaskListener.instanceIsActive();
621+
622+
Field execField = ImportTaskListenerImpl.class.getDeclaredField("executorService");
623+
execField.setAccessible(true);
624+
execField.set(importTaskListener, null);
625+
626+
int threads = 64;
627+
CyclicBarrier start = new CyclicBarrier(threads);
628+
ExecutorService callers = java.util.concurrent.Executors.newFixedThreadPool(threads);
629+
630+
List<Future<ExecutorService>> futures = new ArrayList<>();
631+
for (int i = 0; i < threads; i++) {
632+
futures.add(callers.submit(() -> {
633+
start.await();
634+
return importTaskListener.ensureExecutorAlive();
635+
}));
636+
}
637+
638+
ExecutorService first = null;
639+
for (Future<ExecutorService> f : futures) {
640+
ExecutorService es = f.get(10, TimeUnit.SECONDS);
641+
assertNotNull(es, "Executor should be created");
642+
if (first == null) first = es; else assertSame(first, es, "All callers must see the same instance");
643+
}
644+
645+
callers.shutdownNow();
646+
first.shutdownNow();
647+
}
648+
649+
@Test
650+
public void ensureExecutorAliveRecreatesOnceIfShutdownUnderConcurrency() throws Exception {
651+
// Ensure active mode
652+
importTaskListener.instanceIsActive();
653+
654+
// First creation
655+
ExecutorService first = importTaskListener.ensureExecutorAlive();
656+
assertNotNull(first);
657+
658+
// Force recreate path: mark current as shutdown and ensure the field holds that value
659+
first.shutdown();
660+
661+
Field execField = ImportTaskListenerImpl.class.getDeclaredField("executorService");
662+
execField.setAccessible(true);
663+
execField.set(importTaskListener, first);
664+
665+
int threads = 64;
666+
CyclicBarrier start = new CyclicBarrier(threads);
667+
ExecutorService callers = java.util.concurrent.Executors.newFixedThreadPool(threads);
668+
669+
List<Future<ExecutorService>> futures = new ArrayList<>();
670+
for (int i = 0; i < threads; i++) {
671+
futures.add(callers.submit(() -> {
672+
start.await();
673+
return importTaskListener.ensureExecutorAlive();
674+
}));
675+
}
676+
677+
ExecutorService second = null;
678+
for (Future<ExecutorService> f : futures) {
679+
ExecutorService es = f.get(10, TimeUnit.SECONDS);
680+
assertNotNull(es);
681+
if (second == null) second = es; else assertSame(second, es, "All callers must see the same new instance");
682+
}
683+
684+
assertNotSame(first, second, "Executor must be replaced after shutdown");
685+
callers.shutdownNow();
686+
second.shutdownNow();
687+
}
688+
689+
@Test
690+
public void ensureExecutorAliveReturnsNullWhenPassiveEvenUnderConcurrency() throws Exception {
691+
// Put into passive mode (ensureExecutorAlive should early-return null)
692+
importTaskListener.instanceIsPassive();
693+
694+
Field execField = ImportTaskListenerImpl.class.getDeclaredField("executorService");
695+
execField.setAccessible(true);
696+
execField.set(importTaskListener, null);
697+
698+
int threads = 32;
699+
CyclicBarrier start = new CyclicBarrier(threads);
700+
ExecutorService callers = java.util.concurrent.Executors.newFixedThreadPool(threads);
701+
702+
List<Future<ExecutorService>> futures = new ArrayList<>();
703+
for (int i = 0; i < threads; i++) {
704+
futures.add(callers.submit(() -> {
705+
start.await();
706+
return importTaskListener.ensureExecutorAlive();
707+
}));
708+
}
709+
710+
for (Future<ExecutorService> f : futures) {
711+
assertNull(f.get(5, TimeUnit.SECONDS), "No executor should be created in passive mode");
712+
}
713+
714+
// Field should remain null
715+
assertNull(execField.get(importTaskListener));
716+
callers.shutdownNow();
717+
}
718+
568719
private void setExecutorServiceAndSemaphore(ImportTaskListenerImpl importTaskListener, ExecutorService mockExecutor, Semaphore mockSemaphore) {
569720
try {
570721
Field executorField = ImportTaskListenerImpl.class.getDeclaredField("executorService");

0 commit comments

Comments
 (0)