FrankChen021 commented on code in PR #18525: URL: https://github.com/apache/druid/pull/18525#discussion_r4053225090
########## extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaHeaderBasedFilterConfig.java: ########## @@ -0,0 +1,146 @@ +/* + * 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.druid.indexing.kafka.supervisor; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableSet; +import org.apache.druid.error.InvalidInput; +import org.apache.druid.query.filter.DimFilter; +import org.apache.druid.query.filter.InDimFilter; + +import javax.annotation.Nullable; + +import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; + +/** + * Kafka-specific implementation of header-based filtering. + * Allows filtering Kafka records based on message headers before deserialization. + */ +public class KafkaHeaderBasedFilterConfig +{ + private static final ImmutableSet<Class<? extends DimFilter>> SUPPORTED_FILTER_TYPES = ImmutableSet.of( + InDimFilter.class + ); + + private final DimFilter filter; + private final String encoding; + private final int stringDecodingCacheSize; + + @JsonCreator + public KafkaHeaderBasedFilterConfig( + @JsonProperty("filter") DimFilter filter, + @JsonProperty("encoding") @Nullable String encoding, + @JsonProperty("stringDecodingCacheSize") @Nullable Integer stringDecodingCacheSize + ) + { + this.filter = Preconditions.checkNotNull(filter, "filter cannot be null"); + this.encoding = encoding != null ? encoding : StandardCharsets.UTF_8.name(); + this.stringDecodingCacheSize = stringDecodingCacheSize != null ? stringDecodingCacheSize : 10_000; Review Comment: `stringDecodingCacheSize` isn't validated to be non-negative here, unlike `encoding` and `filter` in the same constructor (which fail cleanly via `IllegalArgumentException`/`InvalidInput.exception`). A spec with e.g. `"stringDecodingCacheSize": -1` is accepted at submission time, but later blows up with an opaque `IllegalArgumentException("maximum size must not be negative")` from `Caffeine.newBuilder().maximumSize(...)` every time a record supplier is constructed (supervisor start, and every task launch) — effectively a crash loop instead of a clean validation error. Please add a non-negative check here. ########## extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaHeaderBasedFilterEvaluator.java: ########## @@ -0,0 +1,184 @@ +/* + * 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.druid.indexing.kafka; + +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; +import org.apache.druid.indexing.kafka.supervisor.KafkaHeaderBasedFilterConfig; +import org.apache.druid.java.util.common.logger.Logger; +import org.apache.druid.query.filter.Filter; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.header.Header; +import org.apache.kafka.common.header.Headers; + +import javax.annotation.Nullable; + +import java.io.UncheckedIOException; +import java.nio.ByteBuffer; +import java.nio.charset.CharacterCodingException; +import java.nio.charset.Charset; +import java.nio.charset.CharsetDecoder; +import java.nio.charset.CodingErrorAction; + +/** + * Evaluates Kafka header filters for pre-ingestion filtering. + */ +public class KafkaHeaderBasedFilterEvaluator +{ + private static final Logger log = new Logger(KafkaHeaderBasedFilterEvaluator.class); + + /** + * Header values larger than this are decoded but not cached, so that a bounded number of cache entries + * (see {@link KafkaHeaderBasedFilterConfig#getStringDecodingCacheSize()}) cannot retain unbounded heap when + * distinct large headers are seen. Total cached bytes are therefore bounded by cacheSize * this cap. + */ + private static final int MAX_CACHEABLE_HEADER_BYTES = 4096; + + private final HeaderFilterHandler filterHandler; + private final String headerName; + private final Charset encoding; + private final Cache<ByteBuffer, String> stringDecodingCache; + + /** + * Creates a new KafkaHeaderBasedFilterEvaluator with the given configuration. + * + * @param headerBasedFilterConfig the configuration containing filter, encoding, and cache settings + * @throws IllegalArgumentException if the filter type is not supported + */ + public KafkaHeaderBasedFilterEvaluator(KafkaHeaderBasedFilterConfig headerBasedFilterConfig) + { + this.encoding = Charset.forName(headerBasedFilterConfig.getEncoding()); + this.stringDecodingCache = Caffeine.newBuilder() + .maximumSize(headerBasedFilterConfig.getStringDecodingCacheSize()) + .build(); + + Filter filter = headerBasedFilterConfig.getFilter().toFilter(); + this.filterHandler = HeaderFilterHandlerFactory.forFilter(filter); + this.headerName = filterHandler.getHeaderName(); + + log.info("Initialized Kafka header filter: %s with encoding [%s] and cache size [%d]", + filterHandler.getDescription(), + headerBasedFilterConfig.getEncoding(), + headerBasedFilterConfig.getStringDecodingCacheSize()); + } + + + /** + * Evaluates whether a Kafka record should be included based on its headers. + * + * @param record the Kafka consumer record + * @return true if the record should be included, false if it should be filtered out + */ + public boolean shouldIncludeRecord(ConsumerRecord<byte[], byte[]> record) + { + try { + return evaluateInclusion(record.headers()); + } + catch (Exception e) { + log.warn( + e, + "Error evaluating header filter for record at topic [%s] partition [%d] offset [%d], including record", + record.topic(), + record.partition(), + record.offset() + ); + return true; // Default to including record on error + } + } + + /** + * Evaluates whether a record should be included based on its headers. + * + * Uses permissive behavior: records with missing, null, or undecodable headers + * are included by default. Only records with successfully decoded header values + * that don't match the filter criteria are excluded. + * + * @param headers the Kafka message headers to evaluate + * @return true if the record should be included, false if it should be filtered out + */ + private boolean evaluateInclusion(Headers headers) + { + // Permissive behavior: missing headers result in inclusion + if (headers == null) { + return true; + } + + Header header = headers.lastHeader(headerName); + + // Permissive behavior: header is missing, or its value is null or zero-length. A zero-length value would + // otherwise decode to an empty string and be matched against the filter, silently dropping such records. + if (header == null || header.value() == null || header.value().length == 0) { + return true; + } + + String headerValue = getDecodedHeaderValue(header.value()); + // Permissive behavior: failed to decode header value + if (headerValue == null) { + return true; + } + + return filterHandler.shouldInclude(headerValue); + } + + + /** + * Decode header bytes to string with caching. + * Returns null if decoding fails, so that undecodable header values fall through to the permissive + * (include the record) path rather than matching on a replacement string. + */ + @Nullable + private String getDecodedHeaderValue(byte[] headerBytes) + { + try { + // Do not cache oversized values so the cache cannot retain unbounded heap for distinct large headers. + if (headerBytes.length > MAX_CACHEABLE_HEADER_BYTES) { + return decodeStrict(headerBytes); + } + ByteBuffer key = ByteBuffer.wrap(headerBytes); + return stringDecodingCache.get(key, k -> decodeStrict(headerBytes)); + } + catch (Exception e) { + // Includes UncheckedIOException wrapping CharacterCodingException for malformed/unmappable input. + // Logged at debug because this is on the per-record path: a high-rate topic with bad bytes or an + // encoding mismatch would otherwise flood logs. The record is still included (permissive behavior). + log.debug(e, "Failed to decode header bytes, treating as null"); + return null; + } + } + + /** + * Decodes bytes with the configured charset, reporting (rather than silently replacing with U+FFFD) malformed + * or unmappable input, so that invalid bytes surface as a failure instead of a bogus match value. + */ + private String decodeStrict(byte[] headerBytes) + { + final CharsetDecoder decoder = encoding.newDecoder() Review Comment: `decodeStrict` allocates a new `CharsetDecoder` on every call (`encoding.newDecoder()...`). This runs on every cache miss and, for headers above `MAX_CACHEABLE_HEADER_BYTES`, on every single record. Since polling is single-threaded per `KafkaRecordSupplier`, this can be a reused instance field with `.reset()` called before each use instead of allocating fresh each time — worth doing given this PR's whole premise is reducing per-record CPU cost on a hot path handling millions of events/sec. ########## extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTaskIOConfig.java: ########## @@ -188,6 +230,14 @@ public boolean isMultiTopic() return multiTopic; } + + @JsonProperty + @Nullable + public KafkaHeaderBasedFilterConfig getheaderBasedFilterConfig() Review Comment: Getter is named `getheaderBasedFilterConfig()` (lowercase `h`) instead of the standard bean casing `getHeaderBasedFilterConfig()` used by every other getter in this class (`getConfigOverrides()`, etc.). JSON round-trips fine today only because the `@JsonCreator` constructor param and Jackson's inferred property name from this getter happen to both resolve to `headerBasedFilterConfig`, but the wrong casing will trip up any caller/tooling using standard naming. Same issue in `KafkaSupervisorIOConfig.getheaderBasedFilterConfig()` (line 239) — please fix both. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
