adnanhemani commented on code in PR #4923: URL: https://github.com/apache/polaris/pull/4923#discussion_r3576122287
########## extensions/events/kafka/src/main/java/org/apache/polaris/extensions/events/kafka/EventToJsonMapper.java: ########## @@ -0,0 +1,117 @@ +/* + * 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.polaris.extensions.events.kafka; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.inject.Inject; +import java.util.HashMap; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.rest.requests.RenameTableRequest; +import org.apache.polaris.service.events.EventAttributes; +import org.apache.polaris.service.events.PolarisEvent; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +@ApplicationScoped +public final class EventToJsonMapper { + private static final Logger LOGGER = LoggerFactory.getLogger(EventToJsonMapper.class); + + private final ObjectMapper mapper; + + @Inject + public EventToJsonMapper(ObjectMapper mapper) { + this.mapper = mapper; + } + + /** + * Extract useful information from PolarisEvent and encode as JSON String. + * + * @param event PolarisEvent to encode. + * @return String containing JSON representation of event details. + * @throws JsonProcessingException + */ + public String eventToJson(PolarisEvent event) throws JsonProcessingException { + HashMap<String, Object> eventProperties = new HashMap<>(); + eventProperties.put("event_type", event.type().name()); + + eventProperties.put("timestamp", event.metadata().timestamp().toEpochMilli()); + event + .attributes() + .get(EventAttributes.RENAME_TABLE_REQUEST) + .map(RenameTableRequest::destination) + .ifPresent(destination -> eventProperties.put("destination", destination.toString())); + event Review Comment: Should we collapse this with the "destination" field above to keep the code DRY? I'd also think that extracting the main logic here into a function would be better as well for DRYness. ########## runtime/defaults/src/main/resources/application.properties: ########## @@ -179,6 +179,12 @@ polaris.file-io.type=default # quarkus.otel.logs.enabled=true # quarkus.otel.exporter.otlp.endpoint=http://otlp-collector:4317 +# Kafka event listener settings Review Comment: Need to add documentation here, similar to: https://polaris.apache.org/releases/1.6.0/configuration/configuration-reference/#polarisevent-listeneraws-cloudwatch ########## extensions/events/kafka/src/main/java/org/apache/polaris/extensions/events/kafka/KafkaEventListenerConfiguration.java: ########## @@ -0,0 +1,65 @@ +/* + * 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.polaris.extensions.events.kafka; + +import io.quarkus.runtime.annotations.StaticInitSafe; +import io.smallrye.config.ConfigMapping; +import io.smallrye.config.WithDefault; +import io.smallrye.config.WithName; +import jakarta.enterprise.context.ApplicationScoped; +import java.util.Map; + +/** Configuration interface for the Kafka Event Listener. */ +@StaticInitSafe +@ConfigMapping(prefix = "polaris.event-listener.kafka") +@ApplicationScoped Review Comment: Nit: `@ApplicationScoped` isn't needed on a `@ConfigMapping` interface — SmallRye/Quarkus already registers config mappings as beans. The sibling `InMemoryBufferEventListenerConfiguration` omits it; suggest dropping it here for consistency. ########## extensions/events/kafka/src/test/java/org/apache/polaris/extensions/events/kafka/KafkaEventListenerTest.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.polaris.extensions.events.kafka; + +import static org.apache.polaris.containerspec.ContainerSpecHelper.containerSpecHelper; +import static org.assertj.core.api.Assertions.as; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.assertj.core.api.InstanceOfAssertFactories.LIST; +import static org.mockito.Mockito.when; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.quarkus.runtime.configuration.MemorySize; +import java.math.BigInteger; +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.Set; +import java.util.UUID; +import java.util.stream.Stream; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.common.serialization.UUIDDeserializer; +import org.apache.polaris.core.auth.PolarisPrincipal; +import org.apache.polaris.service.config.PolarisIcebergObjectMapperCustomizer; +import org.apache.polaris.service.events.EventAttributeMap; +import org.apache.polaris.service.events.EventAttributes; +import org.apache.polaris.service.events.ImmutablePolarisEventMetadata; +import org.apache.polaris.service.events.PolarisEvent; +import org.apache.polaris.service.events.PolarisEventType; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.kafka.KafkaContainer; + +@Testcontainers +@ExtendWith(MockitoExtension.class) +class KafkaEventListenerTest { + @Container + private static KafkaContainer KAFKA_CONTAINER = + new KafkaContainer( + containerSpecHelper("kafka", KafkaEventListenerTest.class) + .dockerImageName(null) + .asCompatibleSubstituteFor("kafka-native")); + + private static final String TOPIC = "polaris-events"; + private static final String REALM = "test-realm"; + private static final String TEST_USER = "test-user"; + private static final String KAFKA_GROUP_ID = "polaris-group-id"; + private static final PolarisPrincipal PRINCIPAL = + PolarisPrincipal.of(TEST_USER, Map.of(), Set.of("role1", "role2")); + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + + static { + new PolarisIcebergObjectMapperCustomizer(new MemorySize(BigInteger.valueOf(1024 * 1024))) + .customize(OBJECT_MAPPER); + } + + @Mock private KafkaEventListenerConfiguration config; + + private KafkaConsumer<UUID, String> getKafkaConsumer() { + Properties props = new Properties(); + props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, UUIDDeserializer.class.getName()); + props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_CONTAINER.getBootstrapServers()); + // Get events from the beginning. + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + props.put(ConsumerConfig.GROUP_ID_CONFIG, KAFKA_GROUP_ID); + return new KafkaConsumer<>(props); + } + + private PolarisEvent getEvent() { + TableIdentifier testTable = TableIdentifier.of("test_namespace", "test_table"); + return new PolarisEvent( + PolarisEventType.AFTER_REFRESH_TABLE, + ImmutablePolarisEventMetadata.builder().realmId(REALM).user(PRINCIPAL).build(), + new EventAttributeMap() + .put(EventAttributes.CATALOG_NAME, "test_catalog") + .put(EventAttributes.TABLE_IDENTIFIER, testTable)); + } + + @Test + void testNoBootStrapServers() { + // Configure the mocks & create listener + when(config.topic()).thenReturn(TOPIC); + // No bootstrap servers. + when(config.properties()).thenReturn(Map.of()); + when(config.synchronousMode()).thenReturn(true); + KafkaEventListener listener = + new KafkaEventListener(config, new EventToJsonMapper(OBJECT_MAPPER)); + // Exception is thrown as Kafka is not configured properly. + assertThatThrownBy( + () -> { + listener.start(); + }) + .isInstanceOf(org.apache.kafka.common.config.ConfigException.class) + .hasMessage( + "Missing required configuration \"bootstrap.servers\" which has no default value."); + } + + @ParameterizedTest + @MethodSource("provideConfigForTest") + void publishEvent(String testname, boolean synchronous) throws JsonProcessingException { + // Start the Kafka container. + KAFKA_CONTAINER.start(); Review Comment: Nit: I believe this explicit `KAFKA_CONTAINER.start()` is redundant — the static `@Container` with `@Testcontainers` already starts the container once before the test class runs (and `start()` is idempotent regardless). ########## extensions/events/kafka/src/test/resources/org/apache/polaris/extensions/events/kafka/Dockerfile-kafka-version: ########## @@ -0,0 +1,22 @@ +# +# 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. +# + +# Dockerfile to provide the image name and tag to a test. +# Version is managed by Renovate - do not edit. +FROM apache/kafka-native:4.3.1 Review Comment: Nit: missing trailing newline at end of file ########## runtime/defaults/src/main/resources/application.properties: ########## @@ -179,6 +179,12 @@ polaris.file-io.type=default # quarkus.otel.logs.enabled=true # quarkus.otel.exporter.otlp.endpoint=http://otlp-collector:4317 +# Kafka event listener settings +# polaris.event-listener.types=kafka Review Comment: Nit: this example correctly uses the current `types` (plural) key, but the PR description still shows the deprecated singular `polaris.event-listener.type=kafka`. Worth updating the description so users don't copy the deprecated form. ########## extensions/events/kafka/src/test/java/org/apache/polaris/extensions/events/kafka/KafkaEventListenerTest.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.polaris.extensions.events.kafka; + +import static org.apache.polaris.containerspec.ContainerSpecHelper.containerSpecHelper; +import static org.assertj.core.api.Assertions.as; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.assertj.core.api.InstanceOfAssertFactories.LIST; +import static org.mockito.Mockito.when; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.quarkus.runtime.configuration.MemorySize; +import java.math.BigInteger; +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.Set; +import java.util.UUID; +import java.util.stream.Stream; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.common.serialization.UUIDDeserializer; +import org.apache.polaris.core.auth.PolarisPrincipal; +import org.apache.polaris.service.config.PolarisIcebergObjectMapperCustomizer; +import org.apache.polaris.service.events.EventAttributeMap; +import org.apache.polaris.service.events.EventAttributes; +import org.apache.polaris.service.events.ImmutablePolarisEventMetadata; +import org.apache.polaris.service.events.PolarisEvent; +import org.apache.polaris.service.events.PolarisEventType; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.kafka.KafkaContainer; + +@Testcontainers +@ExtendWith(MockitoExtension.class) +class KafkaEventListenerTest { + @Container + private static KafkaContainer KAFKA_CONTAINER = Review Comment: Nit: `KAFKA_CONTAINER` is a non-final mutable static. Since its lifecycle is managed by `@Container`/`@Testcontainers`, it can be `final`. ########## runtime/defaults/src/main/resources/application.properties: ########## @@ -179,6 +179,12 @@ polaris.file-io.type=default # quarkus.otel.logs.enabled=true # quarkus.otel.exporter.otlp.endpoint=http://otlp-collector:4317 +# Kafka event listener settings Review Comment: Also missing a `CHANGELOG.md` entry for this new feature. And when documenting (per the comment below), it'd be worth spelling out the published JSON payload schema — note that `EventToJsonMapper` currently only maps a subset of attributes, so for many event types (grants, principal/role, policy, etc.) the payload is sparse (just event_type/timestamp/realm_id/principal/request_id). -- 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]
