FrankChen021 commented on code in PR #18525: URL: https://github.com/apache/druid/pull/18525#discussion_r3674254334
########## extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaHeaderBasedFilterEvaluator.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; + +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.nio.ByteBuffer; +import java.nio.charset.Charset; + +/** + * Evaluates Kafka header filters for pre-ingestion filtering. + */ +public class KafkaHeaderBasedFilterEvaluator +{ + private static final Logger log = new Logger(KafkaHeaderBasedFilterEvaluator.class); + + 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 null or empty + if (header == null || header.value() == null) { Review Comment: [P1] Keep empty and malformed headers on the permissive path The documented permissive contract is not met here. A zero-length value is not caught and decodes to an empty string; additionally, `new String(byte[], Charset)` replaces malformed input rather than throwing, so `getDecodedHeaderValue` does not return null for invalid bytes. Unless the empty or replacement string is explicitly allowed, these records are silently discarded. Check for zero-length values and decode with a `CharsetDecoder` configured with `CodingErrorAction.REPORT`, then include the record on failure. ########## processing/src/main/java/org/apache/druid/segment/incremental/RowIngestionMeters.java: ########## @@ -75,6 +76,9 @@ default long getProcessedBytes() long getThrownAway(); void incrementThrownAway(); + long getFiltered(); Review Comment: [P1] Preserve the RowIngestionMeters extension contract `RowIngestionMeters` is annotated `@ExtensionPoint`, but these new abstract methods make existing external implementations source-incompatible and cause `AbstractMethodError` when the filtered Kafka path calls `incrementFiltered`. Replacing the five-argument `RowIngestionMetersTotals` constructor can likewise cause `NoSuchMethodError` in previously compiled implementations. Provide default zero/no-op methods and retain a five-argument constructor overload delegating to `filtered = 0`. ########## extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaHeaderBasedFilterConfig.java: ########## @@ -0,0 +1,137 @@ +/* + * 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; + + // Validate encoding + try { + Charset.forName(this.encoding); + } + catch (Exception e) { + throw new IllegalArgumentException("Invalid encoding: " + this.encoding, e); + } + + // Validate that only supported filter types are used + validateSupportedFilter(this.filter); + } + + /** + * Validates that the filter is one of the supported types. + * Only 'in' filters are supported for direct evaluation. + */ + private void validateSupportedFilter(DimFilter dimFilter) + { + if (!SUPPORTED_FILTER_TYPES.contains(dimFilter.getClass())) { Review Comment: [P2] Reject unsupported InDimFilter extraction functions This class-only validation accepts an `InDimFilter` with a non-null `extractionFn`, but `InDimFilterHandler` only performs a raw string-set lookup and silently ignores that function. Such a valid serialized Druid filter is evaluated with different semantics and may discard the wrong records. Either reject non-null extraction functions here or apply them in the handler. ########## extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaHeaderBasedFilterEvaluator.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; + +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.nio.ByteBuffer; +import java.nio.charset.Charset; + +/** + * Evaluates Kafka header filters for pre-ingestion filtering. + */ +public class KafkaHeaderBasedFilterEvaluator +{ + private static final Logger log = new Logger(KafkaHeaderBasedFilterEvaluator.class); + + 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()) Review Comment: [P1] Bound the decoding cache by retained bytes `maximumSize` bounds entry count, not retained memory. Each `ByteBuffer` key retains the entire Kafka header byte array, while the value retains its decoded string. With the default 10,000 entries, distinct large headers can retain gigabytes and exhaust a task heap; the documentation even recommends increasing the count for unique values. Use a byte-based `maximumWeight`/weigher and/or skip caching oversized values. ########## extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaRecordSupplier.java: ########## @@ -93,18 +99,38 @@ public KafkaRecordSupplier( boolean multiTopic ) { - this(getKafkaConsumer(sortingMapper, consumerProperties, configOverrides), multiTopic); + this(getKafkaConsumer(sortingMapper, consumerProperties, configOverrides), multiTopic, (KafkaHeaderBasedFilterConfig) null); Review Comment: [P2] Apply the header filter during Kafka sampling `KafkaSamplerSpec#createRecordSupplier` still calls this four-argument overload, which hardcodes a null header filter. Sampling a supervisor spec containing `headerBasedFilterConfig` therefore displays records that actual ingestion tasks discard. Pass `kafkaSupervisorIOConfig.getheaderBasedFilterConfig()` from the sampler. ########## extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaRecordSupplier.java: ########## @@ -172,16 +198,34 @@ public Set<StreamPartition<KafkaTopicPartition>> getAssignment() public List<OrderedPartitionableRecord<KafkaTopicPartition, Long, KafkaRecordEntity>> poll(long timeout) { List<OrderedPartitionableRecord<KafkaTopicPartition, Long, KafkaRecordEntity>> polledRecords = new ArrayList<>(); - for (ConsumerRecord<byte[], byte[]> record : consumer.poll(Duration.ofMillis(timeout))) { - polledRecords.add(new OrderedPartitionableRecord<>( - record.topic(), - new KafkaTopicPartition(multiTopic, record.topic(), record.partition()), - record.offset(), - record.value() == null ? null : ImmutableList.of(new KafkaRecordEntity(record)), - record.timestamp() - )); + for (ConsumerRecord<byte[], byte[]> record : consumer.poll(Duration.ofMillis(timeout))) { + KafkaTopicPartition kafkaPartition = new KafkaTopicPartition(multiTopic, record.topic(), record.partition()); + + // Apply header filter if configured + if (headerFilterEvaluator != null && !headerFilterEvaluator.shouldIncludeRecord(record)) { + // Create filtered record for offset advancement with filtered=true flag + polledRecords.add(new OrderedPartitionableRecord<>( + record.topic(), + kafkaPartition, + record.offset(), + Collections.emptyList(), // Empty list for filtered records Review Comment: [P2] Keep filtered payload bytes in input-byte accounting Replacing the payload with an empty list means `StreamChunkParser` can increment only the filtered-event counter; it never sees `record.value()` and never increments `processedBytes`. Consequently, `ingest/input/bytes` excludes all header-filtered payload bytes, contrary to its documented all-input-bytes contract. Carry the payload byte count on the marker, or retain the entity while bypassing parsing, and increment the byte meter. -- 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]
