You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ranger.apache.org by ra...@apache.org on 2026/06/08 05:13:17 UTC
(ranger) 01/01: RANGER-5633: Wire audit-server Kafka producer and consumer tuning
This is an automated email from the ASF dual-hosted git repository.
ramackri pushed a commit to branch RANGER-5633-patch
in repository https://gitbox.apache.org/repos/asf/ranger.git
commit d8b6fc5a627a5a253b4f84b5237b0f4f6495b22f
Author: ramk <ra...@cloudera.com>
AuthorDate: Mon Jun 8 10:34:59 2026 +0530
RANGER-5633: Wire audit-server Kafka producer and consumer tuning
Wire ingestor producer properties from site XML, apply optional topic
configs at create time, default dispatchers to CooperativeStickyAssignor,
and add unit tests for producer config and topic config wiring.
Co-authored-by: Cursor <cu...@cursor.com>
---
audit-server/audit-common/pom.xml | 6 +
.../ranger/audit/server/AuditServerConstants.java | 34 +++++-
.../ranger/audit/utils/AuditMessageQueueUtils.java | 29 +++++
.../audit/utils/AuditMessageQueueUtilsTest.java | 68 +++++++++++
.../conf/ranger-audit-dispatcher-hdfs-site.xml | 22 ++++
.../conf/ranger-audit-dispatcher-solr-site.xml | 24 ++++
audit-server/audit-ingestor/pom.xml | 6 +
.../ranger/audit/producer/kafka/AuditProducer.java | 128 ++++++++++++++-------
.../resources/conf/ranger-audit-ingestor-site.xml | 68 +++++++++++
.../audit/producer/kafka/AuditProducerTest.java | 106 +++++++++++++++++
10 files changed, 448 insertions(+), 43 deletions(-)
diff --git a/audit-server/audit-common/pom.xml b/audit-server/audit-common/pom.xml
index 73bef80be..5266db1c2 100644
--- a/audit-server/audit-common/pom.xml
+++ b/audit-server/audit-common/pom.xml
@@ -46,6 +46,12 @@
<artifactId>slf4j-api</artifactId>
<version>${slf4j.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <version>${junit.jupiter.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git a/audit-server/audit-common/src/main/java/org/apache/ranger/audit/server/AuditServerConstants.java b/audit-server/audit-common/src/main/java/org/apache/ranger/audit/server/AuditServerConstants.java
index bdfc434a1..58b136b85 100644
--- a/audit-server/audit-common/src/main/java/org/apache/ranger/audit/server/AuditServerConstants.java
+++ b/audit-server/audit-common/src/main/java/org/apache/ranger/audit/server/AuditServerConstants.java
@@ -62,6 +62,38 @@ private AuditServerConstants() {}
public static final String PROP_BUFFER_PARTITIONS = "kafka.topic.partitions.buffer";
public static final String PROP_REPLICATION_FACTOR = "kafka.replication.factor";
+ // Kafka producer tuning (ranger.audit.ingestor.kafka.producer.*)
+ public static final String PROP_KAFKA_PRODUCER_PREFIX = "kafka.producer.";
+ public static final String PROP_PRODUCER_BATCH_SIZE = "batch.size";
+ public static final String PROP_PRODUCER_LINGER_MS = "linger.ms";
+ public static final String PROP_PRODUCER_BUFFER_MEMORY = "buffer.memory";
+ public static final String PROP_PRODUCER_COMPRESSION_TYPE = "compression.type";
+ public static final String PROP_PRODUCER_DELIVERY_TIMEOUT_MS = "delivery.timeout.ms";
+ public static final String PROP_PRODUCER_MAX_REQUEST_SIZE = "max.request.size";
+ public static final String PROP_PRODUCER_MAX_BLOCK_MS = "max.block.ms";
+ public static final String PROP_PRODUCER_BATCH_SEND_TIMEOUT_MS = "batch.send.timeout.ms";
+
+ // Optional topic-level configs applied at topic create (Kafka Admin)
+ public static final String PROP_TOPIC_RETENTION_MS = "kafka.topic.retention.ms";
+ public static final String PROP_TOPIC_COMPRESSION_TYPE = "kafka.topic.compression.type";
+ public static final String PROP_TOPIC_MIN_INSYNC_REPLICAS = "kafka.topic.min.insync.replicas";
+
+ // Kafka producer defaults (high-volume audit JSON; idempotent + acks=all)
+ public static final int DEFAULT_PRODUCER_BATCH_SIZE = 131072; // 128 KiB
+ public static final int DEFAULT_PRODUCER_LINGER_MS = 20;
+ public static final long DEFAULT_PRODUCER_BUFFER_MEMORY = 134217728L; // 128 MiB
+ public static final String DEFAULT_PRODUCER_COMPRESSION_TYPE = "lz4";
+ public static final int DEFAULT_PRODUCER_DELIVERY_TIMEOUT_MS = 120000;
+ public static final int DEFAULT_PRODUCER_MAX_REQUEST_SIZE = 1048576; // 1 MiB
+ public static final int DEFAULT_PRODUCER_MAX_BLOCK_MS = 60000;
+ public static final int DEFAULT_PRODUCER_BATCH_SEND_TIMEOUT_MS = 30000;
+ public static final int DEFAULT_PRODUCER_REQUEST_TIMEOUT_MS = 60000;
+
+ // Optional topic defaults when properties are set (production guidance; RF must allow min ISR)
+ public static final String DEFAULT_TOPIC_RETENTION_MS = "604800000"; // 7 days
+ public static final String DEFAULT_TOPIC_COMPRESSION_TYPE = "lz4";
+ public static final String DEFAULT_TOPIC_MIN_INSYNC_REPLICAS = "2";
+
// Kafka Topic defaults
public static final String DEFAULT_TOPIC = "ranger_audits";
public static final String DEFAULT_SASL_MECHANISM = "PLAIN";
@@ -93,7 +125,7 @@ private AuditServerConstants() {}
AUDIT-SERVER DISPATCHER Configuration
**************************************/
// kafka configuration for the audit-server dispatcher
- public static final String DEFAULT_PARTITION_ASSIGNMENT_STRATEGY = "org.apache.kafka.clients.consumer.RangeAssignor";
+ public static final String DEFAULT_PARTITION_ASSIGNMENT_STRATEGY = "org.apache.kafka.clients.consumer.CooperativeStickyAssignor";
public static final String PROP_DISPATCHER_PREFIX = "ranger.audit.dispatcher";
public static final String PROP_DISPATCHER_THREAD_COUNT = "thread.count";
public static final String PROP_DISPATCHER_OFFSET_COMMIT_STRATEGY = "offset.commit.strategy";
diff --git a/audit-server/audit-common/src/main/java/org/apache/ranger/audit/utils/AuditMessageQueueUtils.java b/audit-server/audit-common/src/main/java/org/apache/ranger/audit/utils/AuditMessageQueueUtils.java
index 8c6085973..bfba27260 100644
--- a/audit-server/audit-common/src/main/java/org/apache/ranger/audit/utils/AuditMessageQueueUtils.java
+++ b/audit-server/audit-common/src/main/java/org/apache/ranger/audit/utils/AuditMessageQueueUtils.java
@@ -85,6 +85,12 @@ public static String createAuditsTopicIfNotExists(Properties props, String propP
LOG.info("Creating topic '{}' with {} partitions and replication factor {}", topicName, partitions, replicationFactor);
NewTopic topic = new NewTopic(topicName, partitions, replicationFactor);
+ Map<String, String> topicConfigs = buildTopicConfigs(props, propPrefix);
+
+ if (topicConfigs != null && !topicConfigs.isEmpty()) {
+ topic.configs(topicConfigs);
+ LOG.info("Topic '{}' configs: {}", topicName, topicConfigs);
+ }
admin.createTopics(Collections.singletonList(topic)).all().get();
@@ -364,4 +370,27 @@ private static int getPartitions(Properties prop, String propPrefix) {
return totalPartitions;
}
+
+ /**
+ * Optional topic-level configs for {@code ranger_audits} at create time.
+ * Properties are applied only when set and non-blank in site XML.
+ */
+ static Map<String, String> buildTopicConfigs(Properties props, String propPrefix) {
+ Map<String, String> configs = new HashMap<>();
+
+ putTopicConfigIfSet(configs, props, propPrefix, AuditServerConstants.PROP_TOPIC_RETENTION_MS, "retention.ms");
+ putTopicConfigIfSet(configs, props, propPrefix, AuditServerConstants.PROP_TOPIC_COMPRESSION_TYPE, "compression.type");
+ putTopicConfigIfSet(configs, props, propPrefix, AuditServerConstants.PROP_TOPIC_MIN_INSYNC_REPLICAS, "min.insync.replicas");
+
+ return configs;
+ }
+
+ private static void putTopicConfigIfSet(Map<String, String> configs, Properties props, String propPrefix,
+ String rangerPropSuffix, String kafkaConfigKey) {
+ String value = MiscUtil.getStringProperty(props, propPrefix + "." + rangerPropSuffix);
+
+ if (value != null && !value.trim().isEmpty()) {
+ configs.put(kafkaConfigKey, value.trim());
+ }
+ }
}
diff --git a/audit-server/audit-common/src/test/java/org/apache/ranger/audit/utils/AuditMessageQueueUtilsTest.java b/audit-server/audit-common/src/test/java/org/apache/ranger/audit/utils/AuditMessageQueueUtilsTest.java
new file mode 100644
index 000000000..e20a69647
--- /dev/null
+++ b/audit-server/audit-common/src/test/java/org/apache/ranger/audit/utils/AuditMessageQueueUtilsTest.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ranger.audit.utils;
+
+import org.apache.ranger.audit.server.AuditServerConstants;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class AuditMessageQueueUtilsTest {
+ private static final String PROP_PREFIX = AuditServerConstants.PROP_PREFIX_AUDIT_SERVER + "kafka";
+
+ @Test
+ public void testBuildTopicConfigsEmptyWhenUnset() {
+ Properties props = new Properties();
+
+ Map<String, String> configs = AuditMessageQueueUtils.buildTopicConfigs(props, PROP_PREFIX);
+
+ assertTrue(configs.isEmpty());
+ }
+
+ @Test
+ public void testBuildTopicConfigsSkipsBlankValues() {
+ Properties props = new Properties();
+ props.setProperty(PROP_PREFIX + "." + AuditServerConstants.PROP_TOPIC_RETENTION_MS, " ");
+ props.setProperty(PROP_PREFIX + "." + AuditServerConstants.PROP_TOPIC_COMPRESSION_TYPE, "lz4");
+ props.setProperty(PROP_PREFIX + "." + AuditServerConstants.PROP_TOPIC_MIN_INSYNC_REPLICAS, "");
+
+ Map<String, String> configs = AuditMessageQueueUtils.buildTopicConfigs(props, PROP_PREFIX);
+
+ assertEquals(1, configs.size());
+ assertEquals("lz4", configs.get("compression.type"));
+ }
+
+ @Test
+ public void testBuildTopicConfigsMapsAllSetProperties() {
+ Properties props = new Properties();
+ props.setProperty(PROP_PREFIX + "." + AuditServerConstants.PROP_TOPIC_RETENTION_MS, "604800000");
+ props.setProperty(PROP_PREFIX + "." + AuditServerConstants.PROP_TOPIC_COMPRESSION_TYPE, " lz4 ");
+ props.setProperty(PROP_PREFIX + "." + AuditServerConstants.PROP_TOPIC_MIN_INSYNC_REPLICAS, "2");
+
+ Map<String, String> configs = AuditMessageQueueUtils.buildTopicConfigs(props, PROP_PREFIX);
+
+ assertEquals(3, configs.size());
+ assertEquals("604800000", configs.get("retention.ms"));
+ assertEquals("lz4", configs.get("compression.type"));
+ assertEquals("2", configs.get("min.insync.replicas"));
+ }
+}
diff --git a/audit-server/audit-dispatcher/dispatcher-hdfs/src/main/resources/conf/ranger-audit-dispatcher-hdfs-site.xml b/audit-server/audit-dispatcher/dispatcher-hdfs/src/main/resources/conf/ranger-audit-dispatcher-hdfs-site.xml
index f7ac647f8..7f1ffc335 100644
--- a/audit-server/audit-dispatcher/dispatcher-hdfs/src/main/resources/conf/ranger-audit-dispatcher-hdfs-site.xml
+++ b/audit-server/audit-dispatcher/dispatcher-hdfs/src/main/resources/conf/ranger-audit-dispatcher-hdfs-site.xml
@@ -90,6 +90,28 @@
<description>Maximum records per poll</description>
</property>
+ <property>
+ <name>ranger.audit.dispatcher.partition.assignment.strategy</name>
+ <value>org.apache.kafka.clients.consumer.CooperativeStickyAssignor</value>
+ <description>Cooperative sticky assignor for K8s pod churn (Kafka 3.9+).</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.session.timeout.ms</name>
+ <value>60000</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.heartbeat.interval.ms</name>
+ <value>10000</value>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.max.poll.interval.ms</name>
+ <value>300000</value>
+ <description>Increase (e.g. 600000) if HDFS per-record writes exceed default 5 min window.</description>
+ </property>
+
<!-- Dispatcher Class (ONLY HDFS) -->
<property>
<name>ranger.audit.dispatcher.class</name>
diff --git a/audit-server/audit-dispatcher/dispatcher-solr/src/main/resources/conf/ranger-audit-dispatcher-solr-site.xml b/audit-server/audit-dispatcher/dispatcher-solr/src/main/resources/conf/ranger-audit-dispatcher-solr-site.xml
index 48ab17e8f..b82128bee 100644
--- a/audit-server/audit-dispatcher/dispatcher-solr/src/main/resources/conf/ranger-audit-dispatcher-solr-site.xml
+++ b/audit-server/audit-dispatcher/dispatcher-solr/src/main/resources/conf/ranger-audit-dispatcher-solr-site.xml
@@ -91,6 +91,30 @@
<description>Maximum records per poll for batch processing</description>
</property>
+ <property>
+ <name>ranger.audit.dispatcher.partition.assignment.strategy</name>
+ <value>org.apache.kafka.clients.consumer.CooperativeStickyAssignor</value>
+ <description>Cooperative sticky assignor for K8s pod churn (Kafka 3.9+). Override with RangeAssignor if needed.</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.session.timeout.ms</name>
+ <value>60000</value>
+ <description>Consumer session timeout (failure detection). Must be > 3 × heartbeat.interval.ms.</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.heartbeat.interval.ms</name>
+ <value>10000</value>
+ <description>Consumer heartbeat interval while idle.</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.dispatcher.max.poll.interval.ms</name>
+ <value>300000</value>
+ <description>Max time between poll() calls while processing a batch (Solr indexing). Increase if batches are slow.</description>
+ </property>
+
<!-- Solr Dispatcher Class -->
<property>
<name>ranger.audit.dispatcher.class</name>
diff --git a/audit-server/audit-ingestor/pom.xml b/audit-server/audit-ingestor/pom.xml
index bcb6d0233..19fa95d9e 100644
--- a/audit-server/audit-ingestor/pom.xml
+++ b/audit-server/audit-ingestor/pom.xml
@@ -388,6 +388,12 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <version>${junit.jupiter.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/AuditProducer.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/AuditProducer.java
index 335c5d1ea..14900caf9 100644
--- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/AuditProducer.java
+++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/AuditProducer.java
@@ -42,6 +42,9 @@
public class AuditProducer implements Runnable {
private static final Logger LOG = LoggerFactory.getLogger(AuditProducer.class);
+ /** Max wait for REST batch send callbacks; overridable via site XML. */
+ private static volatile int batchSendTimeoutMs = AuditServerConstants.DEFAULT_PRODUCER_BATCH_SEND_TIMEOUT_MS;
+
public Properties producerProps = new Properties();
public KafkaProducer<String, String> kafkaProducer;
private volatile boolean running = true;
@@ -49,48 +52,16 @@ public class AuditProducer implements Runnable {
public AuditProducer(Properties props, String propPrefix) throws Exception {
LOG.debug("==> AuditProducer()");
- producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, MiscUtil.getStringProperty(props, propPrefix + "." + AuditServerConstants.PROP_BOOTSTRAP_SERVERS));
- producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, Class.forName("org.apache.kafka.common.serialization.StringSerializer"));
- producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, Class.forName("org.apache.kafka.common.serialization.StringSerializer"));
- producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
- producerProps.put(ProducerConfig.ACKS_CONFIG, "all");
-
- String securityProtocol = MiscUtil.getStringProperty(props, propPrefix + "." + AuditServerConstants.PROP_SECURITY_PROTOCOL, AuditServerConstants.DEFAULT_SECURITY_PROTOCOL);
- producerProps.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, securityProtocol);
-
- producerProps.put(AuditServerConstants.PROP_SASL_MECHANISM, MiscUtil.getStringProperty(props, propPrefix + "." + AuditServerConstants.PROP_SASL_MECHANISM, AuditServerConstants.DEFAULT_SASL_MECHANISM));
- producerProps.put(AuditServerConstants.PROP_SASL_KERBEROS_SERVICE_NAME, AuditServerConstants.DEFAULT_SERVICE_NAME);
-
- if (securityProtocol.toUpperCase().contains(AuditServerConstants.PROP_SECURITY_PROTOCOL_VALUE)) {
- producerProps.put(AuditServerConstants.PROP_SASL_JAAS_CONFIG, AuditMessageQueueUtils.getJAASConfig(props, propPrefix));
- }
-
- producerProps.put(ProducerConfig.LINGER_MS_CONFIG, 5);
- producerProps.put(ProducerConfig.BATCH_SIZE_CONFIG, 32 * 1024);
-
- // Check if configured.plugins is set to determine partitioning strategy
- // 1) configured.plugins is set then Custom AuditPartitioner is used, it allocates predefined set of partitions to each appId.
- // 2) if configured.plugins is not set then default kafka hash based partitioner is used with initial quota of 10 partition.
- String configuredPlugins = MiscUtil.getStringProperty(props, propPrefix + "." + AuditServerConstants.PROP_CONFIGURED_PLUGINS, "");
- if (configuredPlugins != null && !configuredPlugins.trim().isEmpty()) {
- // Plugin-based partitioning: use AuditPartitioner
- String partitionerClass = MiscUtil.getStringProperty(props, propPrefix + "." + AuditServerConstants.PROP_PARTITIONER_CLASS, AuditServerConstants.DEFAULT_PARTITIONER_CLASS);
- producerProps.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, partitionerClass);
- LOG.info("Configured plugins detected - using plugin-based partitioner: {}", partitionerClass);
-
- // Pass all ranger.audit.ingestor.* properties to partitioner (no namespace translation)
- for (String propName : props.stringPropertyNames()) {
- if (propName.startsWith(propPrefix + ".")) {
- producerProps.put(propName, props.getProperty(propName));
- }
- }
- } else {
- // No configured plugins: use Kafka default hash-based partitioner
- LOG.info("No configured plugins - using Kafka default hash-based partitioner");
- }
+ producerProps = createProducerConfig(props, propPrefix);
try {
kafkaProducer = MiscUtil.executePrivilegedAction((PrivilegedExceptionAction<KafkaProducer<String, String>>) () -> new KafkaProducer<>(producerProps));
+ LOG.info("AuditProducer(): KafkaProducer created (batch.size={}, linger.ms={}, compression.type={}, buffer.memory={}, delivery.timeout.ms={})",
+ producerProps.get(ProducerConfig.BATCH_SIZE_CONFIG),
+ producerProps.get(ProducerConfig.LINGER_MS_CONFIG),
+ producerProps.get(ProducerConfig.COMPRESSION_TYPE_CONFIG),
+ producerProps.get(ProducerConfig.BUFFER_MEMORY_CONFIG),
+ producerProps.get(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG));
LOG.info("AuditProducer(): KafkaProducer created successfully!");
} catch (Exception ex) {
LOG.warn("AuditProducer(): Unable to create KafkaProducer - Kafka may not be available. " +
@@ -123,6 +94,7 @@ public void shutdown() {
if (kafkaProducer != null) {
try {
LOG.info("Closing Kafka producer...");
+ kafkaProducer.flush();
kafkaProducer.close();
LOG.info("Kafka producer closed successfully");
} catch (Exception e) {
@@ -137,6 +109,77 @@ public KafkaProducer<String, String> getKafkaProducer() {
return kafkaProducer;
}
+ /**
+ * Build Kafka producer properties from ingestor site XML (testable without connecting to Kafka).
+ */
+ static Properties createProducerConfig(Properties props, String propPrefix) throws Exception {
+ Properties producerProps = new Properties();
+
+ producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, MiscUtil.getStringProperty(props, propPrefix + "." + AuditServerConstants.PROP_BOOTSTRAP_SERVERS));
+ producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, Class.forName("org.apache.kafka.common.serialization.StringSerializer"));
+ producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, Class.forName("org.apache.kafka.common.serialization.StringSerializer"));
+ producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
+ producerProps.put(ProducerConfig.ACKS_CONFIG, "all");
+
+ String producerPrefix = propPrefix + "." + AuditServerConstants.PROP_KAFKA_PRODUCER_PREFIX;
+
+ producerProps.put(ProducerConfig.BATCH_SIZE_CONFIG,
+ MiscUtil.getIntProperty(props, producerPrefix + AuditServerConstants.PROP_PRODUCER_BATCH_SIZE,
+ AuditServerConstants.DEFAULT_PRODUCER_BATCH_SIZE));
+ producerProps.put(ProducerConfig.LINGER_MS_CONFIG,
+ MiscUtil.getIntProperty(props, producerPrefix + AuditServerConstants.PROP_PRODUCER_LINGER_MS,
+ AuditServerConstants.DEFAULT_PRODUCER_LINGER_MS));
+ producerProps.put(ProducerConfig.BUFFER_MEMORY_CONFIG,
+ MiscUtil.getLongProperty(props, producerPrefix + AuditServerConstants.PROP_PRODUCER_BUFFER_MEMORY,
+ AuditServerConstants.DEFAULT_PRODUCER_BUFFER_MEMORY));
+ producerProps.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,
+ MiscUtil.getStringProperty(props, producerPrefix + AuditServerConstants.PROP_PRODUCER_COMPRESSION_TYPE,
+ AuditServerConstants.DEFAULT_PRODUCER_COMPRESSION_TYPE));
+ producerProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG,
+ MiscUtil.getIntProperty(props, producerPrefix + AuditServerConstants.PROP_PRODUCER_DELIVERY_TIMEOUT_MS,
+ AuditServerConstants.DEFAULT_PRODUCER_DELIVERY_TIMEOUT_MS));
+ producerProps.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG,
+ MiscUtil.getIntProperty(props, producerPrefix + AuditServerConstants.PROP_PRODUCER_MAX_REQUEST_SIZE,
+ AuditServerConstants.DEFAULT_PRODUCER_MAX_REQUEST_SIZE));
+ producerProps.put(ProducerConfig.MAX_BLOCK_MS_CONFIG,
+ MiscUtil.getIntProperty(props, producerPrefix + AuditServerConstants.PROP_PRODUCER_MAX_BLOCK_MS,
+ AuditServerConstants.DEFAULT_PRODUCER_MAX_BLOCK_MS));
+ producerProps.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG,
+ MiscUtil.getIntProperty(props, propPrefix + "." + AuditServerConstants.PROP_REQ_TIMEOUT_MS,
+ AuditServerConstants.DEFAULT_PRODUCER_REQUEST_TIMEOUT_MS));
+
+ batchSendTimeoutMs = MiscUtil.getIntProperty(props,
+ producerPrefix + AuditServerConstants.PROP_PRODUCER_BATCH_SEND_TIMEOUT_MS,
+ AuditServerConstants.DEFAULT_PRODUCER_BATCH_SEND_TIMEOUT_MS);
+
+ String securityProtocol = MiscUtil.getStringProperty(props, propPrefix + "." + AuditServerConstants.PROP_SECURITY_PROTOCOL, AuditServerConstants.DEFAULT_SECURITY_PROTOCOL);
+ producerProps.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, securityProtocol);
+
+ producerProps.put(AuditServerConstants.PROP_SASL_MECHANISM, MiscUtil.getStringProperty(props, propPrefix + "." + AuditServerConstants.PROP_SASL_MECHANISM, AuditServerConstants.DEFAULT_SASL_MECHANISM));
+ producerProps.put(AuditServerConstants.PROP_SASL_KERBEROS_SERVICE_NAME, AuditServerConstants.DEFAULT_SERVICE_NAME);
+
+ if (securityProtocol.toUpperCase().contains(AuditServerConstants.PROP_SECURITY_PROTOCOL_VALUE)) {
+ producerProps.put(AuditServerConstants.PROP_SASL_JAAS_CONFIG, AuditMessageQueueUtils.getJAASConfig(props, propPrefix));
+ }
+
+ String configuredPlugins = MiscUtil.getStringProperty(props, propPrefix + "." + AuditServerConstants.PROP_CONFIGURED_PLUGINS, "");
+ if (configuredPlugins != null && !configuredPlugins.trim().isEmpty()) {
+ String partitionerClass = MiscUtil.getStringProperty(props, propPrefix + "." + AuditServerConstants.PROP_PARTITIONER_CLASS, AuditServerConstants.DEFAULT_PARTITIONER_CLASS);
+ producerProps.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, partitionerClass);
+ LOG.info("Configured plugins detected - using plugin-based partitioner: {}", partitionerClass);
+
+ for (String propName : props.stringPropertyNames()) {
+ if (propName.startsWith(propPrefix + ".")) {
+ producerProps.put(propName, props.getProperty(propName));
+ }
+ }
+ } else {
+ LOG.info("No configured plugins - using Kafka default hash-based partitioner");
+ }
+
+ return producerProps;
+ }
+
public static void send(KafkaProducer<String, String> producer, String topic, String key, String value) throws Exception {
ProducerRecord<String, String> auditEvent = new ProducerRecord<>(topic, key, value);
try {
@@ -213,11 +256,12 @@ public void onCompletion(RecordMetadata metadata, Exception e) {
}
}
- // max wait time before timeout
- boolean completed = latch.await(30, TimeUnit.SECONDS);
+ // max wait time before timeout (configurable via ranger.audit.ingestor.kafka.producer.batch.send.timeout.ms)
+ boolean completed = latch.await(batchSendTimeoutMs, TimeUnit.MILLISECONDS);
if (!completed) {
- String errorMsg = String.format("Batch send timed out after 30 seconds. %d/%d events still pending", latch.getCount(), batchSize);
+ String errorMsg = String.format("Batch send timed out after %d ms. %d/%d events still pending",
+ batchSendTimeoutMs, latch.getCount(), batchSize);
LOG.error(errorMsg);
throw new Exception(errorMsg);
}
diff --git a/audit-server/audit-ingestor/src/main/resources/conf/ranger-audit-ingestor-site.xml b/audit-server/audit-ingestor/src/main/resources/conf/ranger-audit-ingestor-site.xml
index 6252a503b..976c2a0cb 100644
--- a/audit-server/audit-ingestor/src/main/resources/conf/ranger-audit-ingestor-site.xml
+++ b/audit-server/audit-ingestor/src/main/resources/conf/ranger-audit-ingestor-site.xml
@@ -374,8 +374,76 @@
<property>
<name>ranger.audit.ingestor.kafka.request.timeout.ms</name>
<value>60000</value>
+ <description>Kafka producer and AdminClient request timeout (ms)</description>
</property>
+ <!-- KAFKA PRODUCER PERFORMANCE (AuditProducer → ProducerConfig) -->
+ <property>
+ <name>ranger.audit.ingestor.kafka.producer.batch.size</name>
+ <value>131072</value>
+ <description>Kafka batch.size in bytes. Default 128 KiB (65536–262144 for high volume).</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.ingestor.kafka.producer.linger.ms</name>
+ <value>20</value>
+ <description>Wait up to this many ms to coalesce records into larger batches (10–50 typical).</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.ingestor.kafka.producer.buffer.memory</name>
+ <value>134217728</value>
+ <description>Producer buffer memory in bytes. Default 128 MiB.</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.ingestor.kafka.producer.compression.type</name>
+ <value>lz4</value>
+ <description>Compression for audit JSON (lz4 recommended: low CPU, good ratio).</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.ingestor.kafka.producer.delivery.timeout.ms</name>
+ <value>120000</value>
+ <description>Upper bound on producer delivery attempts including retries (ms).</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.ingestor.kafka.producer.max.request.size</name>
+ <value>1048576</value>
+ <description>Max size of a produce request; must be less than broker message.max.bytes.</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.ingestor.kafka.producer.max.block.ms</name>
+ <value>60000</value>
+ <description>Max time producer blocks when buffer is full (backpressure).</description>
+ </property>
+
+ <property>
+ <name>ranger.audit.ingestor.kafka.producer.batch.send.timeout.ms</name>
+ <value>30000</value>
+ <description>Application timeout waiting for REST batch send callbacks (ms).</description>
+ </property>
+
+ <!-- Optional topic configs at create time (leave unset for broker defaults; enable for production RF>=3) -->
+ <!--
+ <property>
+ <name>ranger.audit.ingestor.kafka.topic.retention.ms</name>
+ <value>604800000</value>
+ <description>7-day Kafka retention for ranger_audits (replay buffer, not long-term archive)</description>
+ </property>
+ <property>
+ <name>ranger.audit.ingestor.kafka.topic.compression.type</name>
+ <value>lz4</value>
+ </property>
+ <property>
+ <name>ranger.audit.ingestor.kafka.topic.min.insync.replicas</name>
+ <value>2</value>
+ <description>Use only when replication.factor is 3+</description>
+ </property>
+ -->
+
<property>
<name>ranger.audit.ingestor.kafka.connections.max.idle.ms</name>
<value>90000</value>
diff --git a/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/AuditProducerTest.java b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/AuditProducerTest.java
new file mode 100644
index 000000000..356cd6111
--- /dev/null
+++ b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/AuditProducerTest.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ranger.audit.producer.kafka;
+
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.ranger.audit.server.AuditServerConstants;
+import org.junit.jupiter.api.Test;
+
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+public class AuditProducerTest {
+ private static final String PROP_PREFIX = AuditServerConstants.PROP_PREFIX_AUDIT_SERVER + "kafka";
+
+ private static void assertProducerProp(Properties props, String key, Object expected) {
+ assertEquals(String.valueOf(expected), String.valueOf(props.get(key)));
+ }
+
+ @Test
+ public void testCreateProducerConfigDefaults() throws Exception {
+ Properties props = new Properties();
+ props.setProperty(PROP_PREFIX + "." + AuditServerConstants.PROP_BOOTSTRAP_SERVERS, "kafka:9092");
+ props.setProperty(PROP_PREFIX + "." + AuditServerConstants.PROP_SECURITY_PROTOCOL, "PLAINTEXT");
+
+ Properties producerProps = AuditProducer.createProducerConfig(props, PROP_PREFIX);
+
+ assertEquals("kafka:9092", producerProps.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG));
+ assertEquals("true", producerProps.get(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG));
+ assertEquals("all", producerProps.get(ProducerConfig.ACKS_CONFIG));
+ assertProducerProp(producerProps, ProducerConfig.BATCH_SIZE_CONFIG, AuditServerConstants.DEFAULT_PRODUCER_BATCH_SIZE);
+ assertProducerProp(producerProps, ProducerConfig.LINGER_MS_CONFIG, AuditServerConstants.DEFAULT_PRODUCER_LINGER_MS);
+ assertProducerProp(producerProps, ProducerConfig.BUFFER_MEMORY_CONFIG, AuditServerConstants.DEFAULT_PRODUCER_BUFFER_MEMORY);
+ assertEquals(AuditServerConstants.DEFAULT_PRODUCER_COMPRESSION_TYPE,
+ producerProps.get(ProducerConfig.COMPRESSION_TYPE_CONFIG));
+ assertProducerProp(producerProps, ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, AuditServerConstants.DEFAULT_PRODUCER_DELIVERY_TIMEOUT_MS);
+ assertProducerProp(producerProps, ProducerConfig.MAX_REQUEST_SIZE_CONFIG, AuditServerConstants.DEFAULT_PRODUCER_MAX_REQUEST_SIZE);
+ assertProducerProp(producerProps, ProducerConfig.MAX_BLOCK_MS_CONFIG, AuditServerConstants.DEFAULT_PRODUCER_MAX_BLOCK_MS);
+ assertProducerProp(producerProps, ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, AuditServerConstants.DEFAULT_PRODUCER_REQUEST_TIMEOUT_MS);
+ assertEquals("PLAINTEXT", producerProps.get(AdminClientConfig.SECURITY_PROTOCOL_CONFIG));
+ assertFalse(producerProps.containsKey(ProducerConfig.PARTITIONER_CLASS_CONFIG));
+ }
+
+ @Test
+ public void testCreateProducerConfigCustomTuning() throws Exception {
+ Properties props = new Properties();
+ props.setProperty(PROP_PREFIX + "." + AuditServerConstants.PROP_BOOTSTRAP_SERVERS, "broker1:9092,broker2:9092");
+ props.setProperty(PROP_PREFIX + "." + AuditServerConstants.PROP_SECURITY_PROTOCOL, "PLAINTEXT");
+ props.setProperty(PROP_PREFIX + "." + AuditServerConstants.PROP_REQ_TIMEOUT_MS, "45000");
+
+ String producerPrefix = PROP_PREFIX + "." + AuditServerConstants.PROP_KAFKA_PRODUCER_PREFIX;
+ props.setProperty(producerPrefix + AuditServerConstants.PROP_PRODUCER_BATCH_SIZE, "262144");
+ props.setProperty(producerPrefix + AuditServerConstants.PROP_PRODUCER_LINGER_MS, "50");
+ props.setProperty(producerPrefix + AuditServerConstants.PROP_PRODUCER_BUFFER_MEMORY, "268435456");
+ props.setProperty(producerPrefix + AuditServerConstants.PROP_PRODUCER_COMPRESSION_TYPE, "snappy");
+ props.setProperty(producerPrefix + AuditServerConstants.PROP_PRODUCER_DELIVERY_TIMEOUT_MS, "180000");
+ props.setProperty(producerPrefix + AuditServerConstants.PROP_PRODUCER_MAX_REQUEST_SIZE, "2097152");
+ props.setProperty(producerPrefix + AuditServerConstants.PROP_PRODUCER_MAX_BLOCK_MS, "120000");
+
+ Properties producerProps = AuditProducer.createProducerConfig(props, PROP_PREFIX);
+
+ assertProducerProp(producerProps, ProducerConfig.BATCH_SIZE_CONFIG, 262144);
+ assertProducerProp(producerProps, ProducerConfig.LINGER_MS_CONFIG, 50);
+ assertProducerProp(producerProps, ProducerConfig.BUFFER_MEMORY_CONFIG, 268435456L);
+ assertEquals("snappy", producerProps.get(ProducerConfig.COMPRESSION_TYPE_CONFIG));
+ assertProducerProp(producerProps, ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 180000);
+ assertProducerProp(producerProps, ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 2097152);
+ assertProducerProp(producerProps, ProducerConfig.MAX_BLOCK_MS_CONFIG, 120000);
+ assertProducerProp(producerProps, ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 45000);
+ }
+
+ @Test
+ public void testCreateProducerConfigPluginPartitioner() throws Exception {
+ Properties props = new Properties();
+ props.setProperty(PROP_PREFIX + "." + AuditServerConstants.PROP_BOOTSTRAP_SERVERS, "kafka:9092");
+ props.setProperty(PROP_PREFIX + "." + AuditServerConstants.PROP_SECURITY_PROTOCOL, "PLAINTEXT");
+ props.setProperty(PROP_PREFIX + "." + AuditServerConstants.PROP_CONFIGURED_PLUGINS, "hdfs,hive");
+ props.setProperty(PROP_PREFIX + "." + AuditServerConstants.PROP_PARTITIONER_CLASS,
+ AuditServerConstants.DEFAULT_PARTITIONER_CLASS);
+ props.setProperty(PROP_PREFIX + "." + AuditServerConstants.PROP_TOPIC_PARTITIONS_PER_CONFIGURED_PLUGIN, "3");
+
+ Properties producerProps = AuditProducer.createProducerConfig(props, PROP_PREFIX);
+
+ assertEquals(AuditServerConstants.DEFAULT_PARTITIONER_CLASS,
+ producerProps.get(ProducerConfig.PARTITIONER_CLASS_CONFIG));
+ assertEquals("hdfs,hive", producerProps.get(PROP_PREFIX + "." + AuditServerConstants.PROP_CONFIGURED_PLUGINS));
+ assertEquals("3", producerProps.get(PROP_PREFIX + "." + AuditServerConstants.PROP_TOPIC_PARTITIONS_PER_CONFIGURED_PLUGIN));
+ }
+}