Maps changed to ConcurrentHashMaps for SharedExecutorServiceAsyncExecutor

[tinabandaloemruli]

Check List:

• Unit tests: YES
• Documentation: No

changed Maps to ConcurrentHashMaps for SharedExecutorServiceAsyncExecutor

[filiphr]

Thanks for the PR.

Can you please explain why we need to use a ConcurrentHashMap. From what I can see there are 4 places that are iterating the over some of the maps. Only timerJobAcquisitionThreads and timerJobAcquisitionRunnables have their key sets iterated.

We could replace those places with a call to getTenantIds and in that method return a copy of the tenant id in one of the maps.

[StefanHenke]

@filiphr : In our environment (multi tenancy), we are adding new tenants dynamically at runtime which can be done by independent threads. If that happens in parallel for the same tenant, it might lead to several started runnables for the same tenant. Alternatively, we could add a synchronization into the addTenantAsyncExecutor method so that for a given tenant, this cannot be called in parallel.

We had a similar situation some time back in #2393

[filiphr]

Thanks for the explanation @StefanHenke. Knowing your actual problem I do not see how using ConcurrentHashMap will fix the problem.

If we take a look at the addTenantAsyncExecutor method

flowable-engine/modules/flowable-job-service/src/main/java/org/flowable/job/service/impl/asyncexecutor/multitenant/SharedExecutorServiceAsyncExecutor.java

Lines 78 to 99 in b3e26a8

	public void addTenantAsyncExecutor(String tenantId, boolean startExecutor) {
	TenantAwareAcquireTimerJobsRunnable timerRunnable = new TenantAwareAcquireTimerJobsRunnable(this, tenantInfoHolder, tenantId,
	timerLifecycleListener, globalAcquireLockEnabled, globalAcquireLockPrefix, moveTimerExecutorPoolSize);
	timerJobAcquisitionRunnables.put(tenantId, timerRunnable);
	timerJobAcquisitionThreads.put(tenantId, new Thread(timerRunnable));
	TenantAwareAcquireAsyncJobsDueRunnable asyncJobsRunnable = new TenantAwareAcquireAsyncJobsDueRunnable(this, tenantInfoHolder, tenantId,
	asyncJobsDueLifecycleListener, globalAcquireLockEnabled, globalAcquireLockPrefix);
	asyncJobAcquisitionRunnables.put(tenantId, asyncJobsRunnable);
	asyncJobAcquisitionThreads.put(tenantId, new Thread(asyncJobsRunnable));
	TenantAwareResetExpiredJobsRunnable resetExpiredJobsRunnable = new TenantAwareResetExpiredJobsRunnable(this, tenantInfoHolder, tenantId);
	resetExpiredJobsRunnables.put(tenantId, resetExpiredJobsRunnable);
	resetExpiredJobsThreads.put(tenantId, new Thread(resetExpiredJobsRunnable));
	if (startExecutor) {
	startTimerJobAcquisitionForTenant(tenantId);
	startAsyncJobAcquisitionForTenant(tenantId);
	startResetExpiredJobsForTenant(tenantId);
	}
	}

We will see that there is no check whether an entry already exists. Therefore, even if the method is synchronized you can still call it twice with the same tenantId and thus start the multiple runnables for the same tenant.

There are 2 options to actually fix your problem:

1. Adapt the addTenantAsyncExecutor method to remove and stop previous threads
2. Adapt your logic such that you have a syncrhonization before adding new tenants and you will first check if something exists for the tenant before adding it. I think that this option is slightly better since you will be in full control.

e.g.

public void registerTenant(String tenantId) {

    synchornized(monitor) {
        if (!tenantAwareAsyncExecutor.getTenantIds().contains(tenantId)) {
            tenantAwareAsyncExecutor.addTenantAsyncExecutor(tenantId, true);
        }
    }

}

You can of course have a different logic about this, you could stop the threads if you want to, or you could do something else, entirely up to your custom logic.

[StefanHenke]

@filiphr : Thanks for looking into this. You are right. Situation here is a bit different to TenantAwareDataSource . We will manage the synchronization from outside. Unfortunately, I´m not able to close the PR as I was not the original author

[dbmalkovsky]

Closing the PR for @StefanHenke as he requested.