Configurable timeout during ACL synchro for describeAcls call

[AlexisSouquiere]

Problem

We are sometimes gettings a TimeoutException during the ACL synchronisation. It can happen because of a network latency we can have on some clusters.

09:05:14.945 [scheduled-executor-thread-3] ERROR c.m.n.s.e.AccessControlEntryAsyncExecutor - An error occurred collecting ACLs from broker during ACLs synchronization                                                                            │
│ java.util.concurrent.TimeoutException: null                                                                                                                                                                                                       │
│     at java.base/java.util.concurrent.CompletableFuture.timedGet(Unknown Source)                                                                                                                                                                  │
│     at java.base/java.util.concurrent.CompletableFuture.get(Unknown Source)                                                                                                                                                                       │
│     at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:180)                                                                                                                                                            │
│     at com.michelin.ns4kafka.service.executor.AccessControlEntryAsyncExecutor.collectBrokerAcls(AccessControlEntryAsyncExecutor.java:190)                                                                                                         │
│     at com.michelin.ns4kafka.service.executor.AccessControlEntryAsyncExecutor.synchronizeAcls(AccessControlEntryAsyncExecutor.java:73)                                                                                                            │
│     at com.michelin.ns4kafka.service.executor.AccessControlEntryAsyncExecutor.run(AccessControlEntryAsyncExecutor.java:61)                                                                                                                        │
│     at java.base/java.util.ArrayList.forEach(Unknown Source)                                                                                                                                                                                      │
│     at com.michelin.ns4kafka.service.executor.KafkaAsyncExecutorScheduler.schedule(KafkaAsyncExecutorScheduler.java:53)                                                                                                                           │
│     at com.michelin.ns4kafka.service.executor.$KafkaAsyncExecutorScheduler$Definition$Exec.dispatch(Unknown Source)                                                                                                                               │
│     at io.micronaut.context.AbstractExecutableMethodsDefinition$DispatchedExecutableMethod.invoke(AbstractExecutableMethodsDefinition.java:456)                                                                                                   │
│     at io.micronaut.inject.DelegatingExecutableMethod.invoke(DelegatingExecutableMethod.java:86)                                                                                                                                                  │
│     at io.micronaut.context.bind.DefaultExecutableBeanContextBinder$ContextBoundExecutable.invoke(DefaultExecutableBeanContextBinder.java:152)                                                                                                    │
│     at io.micronaut.scheduling.processor.ScheduledMethodProcessor.lambda$process$2(ScheduledMethodProcessor.java:131)                                                                                                                             │
│     at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)                                                                                                                                                              │
│     at java.base/java.util.concurrent.FutureTask.runAndReset(Unknown Source)                                                                                                                                                                      │
│     at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(Unknown Source)                                                                                                                                         │
│     at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)                                                                                                                                                                │
│     at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)                                                                                                                                                               │
│     at java.base/java.lang.Thread.run(Unknown Source)                                    

Suggestion

The timeout value is fixed (10s). We should be able to change this timeout value through a new configuration parameter.

ns4kafka/src/main/java/com/michelin/ns4kafka/service/executor/AccessControlEntryAsyncExecutor.java

Lines 188 to 193 in 0d0c2b9

	List<AclBinding> userAcls = getAdminClient()
	.describeAcls(aclBindingFilter)
	.values().get(10, TimeUnit.SECONDS)
	.stream()
	.filter(aclBinding -> validResourceTypes.contains(aclBinding.pattern().resourceType()))
	.toList();

Alternatives Considered

N/A

Additional Context

N/A

[loicgreffier]

@AlexisSouquiere The PR will bring configuration of all AdminClient timeouts:

ns4kafka:
  managed-clusters:
    local:
      timeout:
        acl:
          create: 15000
          delete: 15000
          describe: 15000
        topic:
          alter-configs: 10000
          create: 10000
          describe-configs: 30000
          delete: 30000
          list: 30000
        user:
          alter-client-quotas: 10000
          alter-scram-credentials: 10000
          describe-quotas: 10000