You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@ranger.apache.org by "rameeshm (via GitHub)" <gi...@apache.org> on 2026/09/11 18:54:24 UTC
[PR] RANGER-5752:Audit dispatcher consumer dies on TopicAuthorizationException - container stays healthy, audits stop flowing [ranger]
rameeshm opened a new pull request, #1231:
URL: https://github.com/apache/ranger/pull/1231
## What changes were proposed in this pull request?
The patch addresses the issue where the AuditDispatcherBase.DispatcherWorker dies when Kafka is not ready yet.
Retry Logic:
It wraps the poll() and processing logic in a try-catch block. When an AuthorizationException or AuthenticationException is thrown, the worker now sleeps for a configurable delay (auth.retry.delay.ms) and continues the loop instead of exiting.
Worker Health Monitoring:
It adds monitorAndRestartWorkers() to the main thread's loop. If a worker thread terminates unexpectedly (e.g., due to an unhandled Error), the main thread will detect that its Future is done and restart the worker.
## How was this patch tested?
Patch tested in docker with scenarios to repro the situation and see how resilient is AuditDispatcher is.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@ranger.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RANGER-5752:Audit dispatcher consumer dies on TopicAuthorizationException - container stays healthy, audits stop flowing [ranger]
Posted by "maheshbandal15 (via GitHub)" <gi...@apache.org>.
maheshbandal15 commented on code in PR #1231:
URL: https://github.com/apache/ranger/pull/1231#discussion_r4148500091
##########
audit-server/audit-dispatcher/dispatcher-common/src/main/java/org/apache/ranger/audit/dispatcher/kafka/AuditDispatcherBase.java:
##########
@@ -102,10 +121,13 @@ public AuditDispatcherBase(Properties props, String propPrefix, String dispatche
topicName = MiscUtil.getStringProperty(props, propPrefix + "." + AuditServerConstants.PROP_TOPIC_NAME, AuditServerConstants.DEFAULT_TOPIC);
}
- AuditDispatcherBase(String dispatcherGroupId, KafkaConsumer<String, String> dispatcher, String topicName) {
- this.dispatcherGroupId = dispatcherGroupId;
- this.dispatcher = dispatcher;
- this.topicName = topicName;
+ AuditDispatcherBase(String dispatcherGroupId, KafkaConsumer<String, String> dispatcher, String topicName, long authzRetryDelayMs, long authnRetryDelayMs, long pollErrorRetryDelayMs) {
+ this.dispatcherGroupId = dispatcherGroupId;
+ this.dispatcher = dispatcher;
+ this.topicName = topicName;
+ this.authzRetryDelayMs = authzRetryDelayMs;
+ this.authnRetryDelayMs = authnRetryDelayMs;
+ this.pollErrorRetryDelayMs = pollErrorRetryDelayMs;
}
Review Comment:
@rameeshm AuditDispatcherBase.java now exposes a 6-arg constructor but [AuditOpenSearchDispatcher.java#L56](https://github.com/apache/ranger/blob/63e5bd7040954455f89c5f6f4615a92d18093510/audit-server/audit-dispatcher/dispatcher-opensearch/src/main/java/org/apache/ranger/audit/dispatcher/kafka/AuditOpenSearchDispatcher.java#L56) still calls `super(dispatcherGroupId, dispatcher, topicName);` which results in compilation failure.
run `cd audit-server && mvn clean install`
> ranger/audit-server/audit-dispatcher/dispatcher-opensearch/src/main/java/org/apache/ranger/audit/dispatcher/kafka/AuditOpenSearchDispatcher.java:[56,14] error: incompatible types: String cannot be converted to Properties
Please consider restoring 3-arg constructor or update AuditOpenSearchDispatcher’s constructor to pass the same delay values as the base tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@ranger.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RANGER-5752:Audit dispatcher consumer dies on TopicAuthorizationException - container stays healthy, audits stop flowing [ranger]
Posted by "mneethiraj (via GitHub)" <gi...@apache.org>.
mneethiraj commented on code in PR #1231:
URL: https://github.com/apache/ranger/pull/1231#discussion_r3993975992
##########
audit-server/audit-common/src/main/java/org/apache/ranger/audit/server/AuditServerConstants.java:
##########
@@ -135,11 +135,15 @@ private AuditServerConstants() {}
public static final String PROP_DISPATCHER_MAX_POLL_INTERVAL_MS = "max.poll.interval.ms";
public static final String PROP_DISPATCHER_HEARTBEAT_INTERVAL_MS = "heartbeat.interval.ms";
public static final String PROP_DISPATCHER_PARTITION_ASSIGNMENT_STRATEGY = "partition.assignment.strategy";
+ public static final String PROP_DISPATCHER_AUTH_RETRY_DELAY_MS = "auth.retry.delay.ms";
Review Comment:
I suggest replacing "auth" with "authz" - to make it clear this configuration is about retry on authorization failure (instead of authentication failure).
`PROP_DISPATCHER_AUTH_RETRY_DELAY_MS` => `PROP_DISPATCHER_AUTHZ_RETRY_DELAY_MS`
`auth.retry.delay.ms` => `authz.retry.delay.ms`
##########
audit-server/audit-dispatcher/dispatcher-common/src/main/java/org/apache/ranger/audit/dispatcher/kafka/AuditDispatcherBase.java:
##########
@@ -54,10 +59,13 @@ public abstract class AuditDispatcherBase implements AuditDispatcher {
protected final AtomicBoolean running = new AtomicBoolean(false);
protected final Map<String, DispatcherWorker> dispatcherWorkers = new ConcurrentHashMap<>();
+ protected final Map<String, Future<?>> workerFutures = new ConcurrentHashMap<>();
protected ExecutorService dispatcherThreadPool;
protected int dispatcherThreadCount = 1;
protected String offsetCommitStrategy = AuditServerConstants.DEFAULT_OFFSET_COMMIT_STRATEGY;
protected long offsetCommitInterval = AuditServerConstants.DEFAULT_OFFSET_COMMIT_INTERVAL_MS;
+ protected long authRetryDelayMs = AuditServerConstants.DEFAULT_DISPATCHER_AUTH_RETRY_DELAY_MS;
Review Comment:
Consider marking members `authRetryDelayMs` and `pollErrorRetryDelayMs` as `final`, and initializing them only in the constructor at lines 105 and 106 below.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@ranger.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RANGER-5752:Audit dispatcher consumer dies on TopicAuthorizationException - container stays healthy, audits stop flowing [ranger]
Posted by "kumaab (via GitHub)" <gi...@apache.org>.
kumaab commented on code in PR #1231:
URL: https://github.com/apache/ranger/pull/1231#discussion_r3992621771
##########
audit-server/audit-dispatcher/dispatcher-common/src/main/java/org/apache/ranger/audit/dispatcher/kafka/AuditDispatcherBase.java:
##########
@@ -205,16 +218,44 @@ protected void startDispatcherWorkers() {
for (int i = 0; i < dispatcherThreadCount; i++) {
String workerId = getDispatcherName().toLowerCase() + "-worker-" + i;
- DispatcherWorker worker = createDispatcherWorker(workerId, new ArrayList<>());
- dispatcherWorkers.put(workerId, worker);
- dispatcherThreadPool.submit(worker);
-
- LOG.info("Started {} dispatcher worker '{}' - will process ANY appId assigned by Kafka", getDispatcherName(), workerId);
+ startWorker(workerId);
}
LOG.info("<== AuditDispatcherBase.startDispatcherWorkers(): All {} workers started in SUBSCRIBE mode", dispatcherThreadCount);
}
+ private void startWorker(String workerId) {
+ DispatcherWorker worker = createDispatcherWorker(workerId, new ArrayList<>());
+ dispatcherWorkers.put(workerId, worker);
+ Future<?> future = dispatcherThreadPool.submit(worker);
+ workerFutures.put(workerId, future);
+ LOG.info("Started {} dispatcher worker '{}' - will process ANY appId assigned by Kafka", getDispatcherName(), workerId);
+ }
+
+ private void monitorAndRestartWorkers() {
+ if (!running.get()) {
+ return;
+ }
+
+ for (Map.Entry<String, Future<?>> entry : workerFutures.entrySet()) {
+ String workerId = entry.getKey();
+ Future<?> future = entry.getValue();
+
+ if (future.isDone()) {
+ LOG.warn("Worker '{}' has terminated unexpectedly. Attempting to restart...", workerId);
+ try {
+ // Remove the dead worker
+ dispatcherWorkers.remove(workerId);
+ // Start a new worker with the same ID
+ startWorker(workerId);
+ LOG.info("Successfully restarted worker '{}'", workerId);
Review Comment:
It may be helpful to add the number of restarts happened so far, to help with dianostics.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@ranger.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RANGER-5752:Audit dispatcher consumer dies on TopicAuthorizationException - container stays healthy, audits stop flowing [ranger]
Posted by "kumaab (via GitHub)" <gi...@apache.org>.
kumaab commented on code in PR #1231:
URL: https://github.com/apache/ranger/pull/1231#discussion_r3992621771
##########
audit-server/audit-dispatcher/dispatcher-common/src/main/java/org/apache/ranger/audit/dispatcher/kafka/AuditDispatcherBase.java:
##########
@@ -205,16 +218,44 @@ protected void startDispatcherWorkers() {
for (int i = 0; i < dispatcherThreadCount; i++) {
String workerId = getDispatcherName().toLowerCase() + "-worker-" + i;
- DispatcherWorker worker = createDispatcherWorker(workerId, new ArrayList<>());
- dispatcherWorkers.put(workerId, worker);
- dispatcherThreadPool.submit(worker);
-
- LOG.info("Started {} dispatcher worker '{}' - will process ANY appId assigned by Kafka", getDispatcherName(), workerId);
+ startWorker(workerId);
}
LOG.info("<== AuditDispatcherBase.startDispatcherWorkers(): All {} workers started in SUBSCRIBE mode", dispatcherThreadCount);
}
+ private void startWorker(String workerId) {
+ DispatcherWorker worker = createDispatcherWorker(workerId, new ArrayList<>());
+ dispatcherWorkers.put(workerId, worker);
+ Future<?> future = dispatcherThreadPool.submit(worker);
+ workerFutures.put(workerId, future);
+ LOG.info("Started {} dispatcher worker '{}' - will process ANY appId assigned by Kafka", getDispatcherName(), workerId);
+ }
+
+ private void monitorAndRestartWorkers() {
+ if (!running.get()) {
+ return;
+ }
+
+ for (Map.Entry<String, Future<?>> entry : workerFutures.entrySet()) {
+ String workerId = entry.getKey();
+ Future<?> future = entry.getValue();
+
+ if (future.isDone()) {
+ LOG.warn("Worker '{}' has terminated unexpectedly. Attempting to restart...", workerId);
+ try {
+ // Remove the dead worker
+ dispatcherWorkers.remove(workerId);
+ // Start a new worker with the same ID
+ startWorker(workerId);
+ LOG.info("Successfully restarted worker '{}'", workerId);
Review Comment:
It may be helpful to add the number of restarts happened so far, to help with diagnostics.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@ranger.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RANGER-5752:Audit dispatcher consumer dies on TopicAuthorizationException - container stays healthy, audits stop flowing [ranger]
Posted by "rameeshm (via GitHub)" <gi...@apache.org>.
rameeshm commented on code in PR #1231:
URL: https://github.com/apache/ranger/pull/1231#discussion_r4151137560
##########
audit-server/audit-dispatcher/dispatcher-common/src/main/java/org/apache/ranger/audit/dispatcher/kafka/AuditDispatcherBase.java:
##########
@@ -102,10 +121,13 @@ public AuditDispatcherBase(Properties props, String propPrefix, String dispatche
topicName = MiscUtil.getStringProperty(props, propPrefix + "." + AuditServerConstants.PROP_TOPIC_NAME, AuditServerConstants.DEFAULT_TOPIC);
}
- AuditDispatcherBase(String dispatcherGroupId, KafkaConsumer<String, String> dispatcher, String topicName) {
- this.dispatcherGroupId = dispatcherGroupId;
- this.dispatcher = dispatcher;
- this.topicName = topicName;
+ AuditDispatcherBase(String dispatcherGroupId, KafkaConsumer<String, String> dispatcher, String topicName, long authzRetryDelayMs, long authnRetryDelayMs, long pollErrorRetryDelayMs) {
+ this.dispatcherGroupId = dispatcherGroupId;
+ this.dispatcher = dispatcher;
+ this.topicName = topicName;
+ this.authzRetryDelayMs = authzRetryDelayMs;
+ this.authnRetryDelayMs = authnRetryDelayMs;
+ this.pollErrorRetryDelayMs = pollErrorRetryDelayMs;
}
Review Comment:
@maheshbandal15 Thanks for the review. I have update to have only 3 args and pass the values for testing and won't have any separate constructor.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@ranger.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RANGER-5752:Audit dispatcher consumer dies on TopicAuthorizationException - container stays healthy, audits stop flowing [ranger]
Posted by "mneethiraj (via GitHub)" <gi...@apache.org>.
mneethiraj commented on code in PR #1231:
URL: https://github.com/apache/ranger/pull/1231#discussion_r4353957784
##########
audit-server/audit-dispatcher/dispatcher-common/src/main/java/org/apache/ranger/audit/dispatcher/kafka/AuditDispatcherBase.java:
##########
@@ -301,6 +355,44 @@ public void run() {
}
}
+ private boolean handlePollException(Exception e) {
+ if (e instanceof WakeupException) {
+ if (!running.get()) {
+ LOG.info("WakeupException in {} dispatcher worker '{}' during shutdown", getDispatcherName(), workerId);
+ return false;
+ }
+ LOG.error("WakeupException in {} dispatcher worker '{}'", getDispatcherName(), workerId, e);
+ return true;
+ } else if (e instanceof InterruptException) {
+ if (!running.get()) {
+ LOG.info("InterruptException in {} dispatcher worker '{}' during shutdown", getDispatcherName(), workerId);
+ return false;
Review Comment:
Shouldn't `Thread.currentThread().interrupt();` be called from here as well, similar to line 372? If yes, consider following update:
```
if (!running.get()) {
LOG.info("InterruptException in {} dispatcher worker '{}' during shutdown", getDispatcherName(), workerId);
} else {
LOG.error("InterruptException in {} dispatcher worker '{}'", getDispatcherName(), workerId, e);
}
Thread.currentThread().interrupt();
return false;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@ranger.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org