github-advanced-security[bot] commented on code in PR #19311: URL: https://github.com/apache/druid/pull/19311#discussion_r3210682966
########## extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaShareGroupRecordSupplierTest.java: ########## @@ -0,0 +1,300 @@ +/* + * 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 org.apache.druid.data.input.kafka.KafkaRecordEntity; +import org.apache.druid.data.input.kafka.KafkaTopicPartition; +import org.apache.druid.indexing.seekablestream.common.AcknowledgeType; +import org.apache.druid.indexing.seekablestream.common.OrderedPartitionableRecord; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaShareConsumer; +import org.apache.kafka.common.KafkaException; +import org.apache.kafka.common.TopicIdPartition; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.Uuid; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mockito; + +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +public class KafkaShareGroupRecordSupplierTest +{ + private KafkaShareConsumer<byte[], byte[]> mockConsumer; + private KafkaShareGroupRecordSupplier supplier; + + @SuppressWarnings("unchecked") + @Before + public void setUp() + { + mockConsumer = Mockito.mock(KafkaShareConsumer.class); + supplier = new KafkaShareGroupRecordSupplier(mockConsumer); + } + + @After + public void tearDown() + { + supplier.close(); + } + + @Test + public void testSubscribeAndSubscription() + { + final Set<String> topics = Set.of("topic-a", "topic-b"); + Mockito.when(mockConsumer.subscription()).thenReturn(topics); + + supplier.subscribe(topics); + Mockito.verify(mockConsumer).subscribe(topics); + Assert.assertEquals(topics, supplier.subscription()); + } + + @Test + public void testUnsubscribe() + { + supplier.unsubscribe(); + Mockito.verify(mockConsumer).unsubscribe(); + } + + @Test + public void testPollWrapsRecords() + { + final ConsumerRecord<byte[], byte[]> record1 = new ConsumerRecord<>( + "test-topic", 0, 100L, "key1".getBytes(StandardCharsets.UTF_8), "value1".getBytes(StandardCharsets.UTF_8) + ); + final ConsumerRecord<byte[], byte[]> record2 = new ConsumerRecord<>( + "test-topic", 1, 200L, "key2".getBytes(StandardCharsets.UTF_8), "value2".getBytes(StandardCharsets.UTF_8) + ); + + final Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> recordMap = new HashMap<>(); + recordMap.put(new TopicPartition("test-topic", 0), List.of(record1)); + recordMap.put(new TopicPartition("test-topic", 1), List.of(record2)); + final ConsumerRecords<byte[], byte[]> consumerRecords = new ConsumerRecords<>(recordMap); Review Comment: ## CodeQL / Deprecated method or constructor invocation Invoking [ConsumerRecords.ConsumerRecords](1) should be avoided because it has been deprecated. [Show more details](https://github.com/apache/druid/security/code-scanning/11185) ########## extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaShareGroupRecordSupplierTest.java: ########## @@ -0,0 +1,300 @@ +/* + * 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 org.apache.druid.data.input.kafka.KafkaRecordEntity; +import org.apache.druid.data.input.kafka.KafkaTopicPartition; +import org.apache.druid.indexing.seekablestream.common.AcknowledgeType; +import org.apache.druid.indexing.seekablestream.common.OrderedPartitionableRecord; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaShareConsumer; +import org.apache.kafka.common.KafkaException; +import org.apache.kafka.common.TopicIdPartition; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.Uuid; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mockito; + +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +public class KafkaShareGroupRecordSupplierTest +{ + private KafkaShareConsumer<byte[], byte[]> mockConsumer; + private KafkaShareGroupRecordSupplier supplier; + + @SuppressWarnings("unchecked") + @Before + public void setUp() + { + mockConsumer = Mockito.mock(KafkaShareConsumer.class); + supplier = new KafkaShareGroupRecordSupplier(mockConsumer); + } + + @After + public void tearDown() + { + supplier.close(); + } + + @Test + public void testSubscribeAndSubscription() + { + final Set<String> topics = Set.of("topic-a", "topic-b"); + Mockito.when(mockConsumer.subscription()).thenReturn(topics); + + supplier.subscribe(topics); + Mockito.verify(mockConsumer).subscribe(topics); + Assert.assertEquals(topics, supplier.subscription()); + } + + @Test + public void testUnsubscribe() + { + supplier.unsubscribe(); + Mockito.verify(mockConsumer).unsubscribe(); + } + + @Test + public void testPollWrapsRecords() + { + final ConsumerRecord<byte[], byte[]> record1 = new ConsumerRecord<>( + "test-topic", 0, 100L, "key1".getBytes(StandardCharsets.UTF_8), "value1".getBytes(StandardCharsets.UTF_8) + ); + final ConsumerRecord<byte[], byte[]> record2 = new ConsumerRecord<>( + "test-topic", 1, 200L, "key2".getBytes(StandardCharsets.UTF_8), "value2".getBytes(StandardCharsets.UTF_8) + ); + + final Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> recordMap = new HashMap<>(); + recordMap.put(new TopicPartition("test-topic", 0), List.of(record1)); + recordMap.put(new TopicPartition("test-topic", 1), List.of(record2)); + final ConsumerRecords<byte[], byte[]> consumerRecords = new ConsumerRecords<>(recordMap); + + Mockito.when(mockConsumer.poll(Mockito.any(Duration.class))).thenReturn(consumerRecords); + + final List<OrderedPartitionableRecord<KafkaTopicPartition, Long, KafkaRecordEntity>> result = + supplier.poll(1000); + + Assert.assertEquals(2, result.size()); + + final OrderedPartitionableRecord<KafkaTopicPartition, Long, KafkaRecordEntity> polled1 = + result.stream().filter(r -> r.getSequenceNumber() == 100L).findFirst().orElse(null); + Assert.assertNotNull(polled1); + Assert.assertEquals("test-topic", polled1.getStream()); + Assert.assertEquals(0, polled1.getPartitionId().partition()); + Assert.assertNotNull(polled1.getData()); + Assert.assertEquals(1, polled1.getData().size()); + } + + @Test + public void testPollReturnsEmptyOnTimeout() + { + Mockito.when(mockConsumer.poll(Mockito.any(Duration.class))).thenReturn(ConsumerRecords.empty()); + final List<OrderedPartitionableRecord<KafkaTopicPartition, Long, KafkaRecordEntity>> result = + supplier.poll(100); + Assert.assertTrue(result.isEmpty()); + } + + @Test + public void testAcknowledgeDefaultAccept() + { + final ConsumerRecord<byte[], byte[]> record = new ConsumerRecord<>( + "test-topic", 0, 42L, "key".getBytes(StandardCharsets.UTF_8), "value".getBytes(StandardCharsets.UTF_8) + ); + pollSingleRecord(record); + + final KafkaTopicPartition partition = new KafkaTopicPartition(true, "test-topic", 0); + supplier.acknowledge(partition, 42L); + + Mockito.verify(mockConsumer).acknowledge( + Mockito.same(record), + Mockito.eq(org.apache.kafka.clients.consumer.AcknowledgeType.ACCEPT) + ); + } + + @Test + public void testAcknowledgeWithRelease() + { + final ConsumerRecord<byte[], byte[]> record = new ConsumerRecord<>( + "test-topic", 0, 10L, "key".getBytes(StandardCharsets.UTF_8), "value".getBytes(StandardCharsets.UTF_8) + ); + pollSingleRecord(record); + + final KafkaTopicPartition partition = new KafkaTopicPartition(true, "test-topic", 0); + supplier.acknowledge(partition, 10L, AcknowledgeType.RELEASE); + + Mockito.verify(mockConsumer).acknowledge( + Mockito.same(record), + Mockito.eq(org.apache.kafka.clients.consumer.AcknowledgeType.RELEASE) + ); + } + + @Test + public void testAcknowledgeWithReject() + { + final ConsumerRecord<byte[], byte[]> record = new ConsumerRecord<>( + "test-topic", 0, 10L, "key".getBytes(StandardCharsets.UTF_8), "value".getBytes(StandardCharsets.UTF_8) + ); + pollSingleRecord(record); + + final KafkaTopicPartition partition = new KafkaTopicPartition(true, "test-topic", 0); + supplier.acknowledge(partition, 10L, AcknowledgeType.REJECT); + + Mockito.verify(mockConsumer).acknowledge( + Mockito.same(record), + Mockito.eq(org.apache.kafka.clients.consumer.AcknowledgeType.REJECT) + ); + } + + @Test + public void testAcknowledgeWithRenew() + { + final ConsumerRecord<byte[], byte[]> record = new ConsumerRecord<>( + "test-topic", 0, 99L, "key".getBytes(StandardCharsets.UTF_8), "value".getBytes(StandardCharsets.UTF_8) + ); + pollSingleRecord(record); + + final KafkaTopicPartition partition = new KafkaTopicPartition(true, "test-topic", 0); + supplier.acknowledge(partition, 99L, AcknowledgeType.RENEW); + + Mockito.verify(mockConsumer).acknowledge( + Mockito.same(record), + Mockito.eq(org.apache.kafka.clients.consumer.AcknowledgeType.RENEW) + ); + } + + @Test + public void testAcknowledgeBatch() + { + final List<ConsumerRecord<byte[], byte[]>> recordsP0 = Arrays.asList( + new ConsumerRecord<>("test-topic", 0, 1L, "k".getBytes(StandardCharsets.UTF_8), "v".getBytes(StandardCharsets.UTF_8)), + new ConsumerRecord<>("test-topic", 0, 2L, "k".getBytes(StandardCharsets.UTF_8), "v".getBytes(StandardCharsets.UTF_8)), + new ConsumerRecord<>("test-topic", 0, 3L, "k".getBytes(StandardCharsets.UTF_8), "v".getBytes(StandardCharsets.UTF_8)) + ); + final List<ConsumerRecord<byte[], byte[]>> recordsP1 = Arrays.asList( + new ConsumerRecord<>("test-topic", 1, 10L, "k".getBytes(StandardCharsets.UTF_8), "v".getBytes(StandardCharsets.UTF_8)), + new ConsumerRecord<>("test-topic", 1, 11L, "k".getBytes(StandardCharsets.UTF_8), "v".getBytes(StandardCharsets.UTF_8)) + ); + final Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> recordMap = new HashMap<>(); + recordMap.put(new TopicPartition("test-topic", 0), recordsP0); + recordMap.put(new TopicPartition("test-topic", 1), recordsP1); + Mockito.when(mockConsumer.poll(Mockito.any(Duration.class))) + .thenReturn(new ConsumerRecords<>(recordMap)); + supplier.poll(1000); + + final KafkaTopicPartition p0 = new KafkaTopicPartition(true, "test-topic", 0); + final KafkaTopicPartition p1 = new KafkaTopicPartition(true, "test-topic", 1); + + final Map<KafkaTopicPartition, java.util.Collection<Long>> offsets = new HashMap<>(); + offsets.put(p0, Arrays.asList(1L, 2L, 3L)); + offsets.put(p1, Arrays.asList(10L, 11L)); + + supplier.acknowledge(offsets, AcknowledgeType.ACCEPT); + + Mockito.verify(mockConsumer, Mockito.times(5)).acknowledge( + Mockito.<ConsumerRecord<byte[], byte[]>>any(), + Mockito.eq(org.apache.kafka.clients.consumer.AcknowledgeType.ACCEPT) + ); + } + + private void pollSingleRecord(ConsumerRecord<byte[], byte[]> record) + { + final Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> recordMap = new HashMap<>(); + recordMap.put(new TopicPartition(record.topic(), record.partition()), List.of(record)); + Mockito.when(mockConsumer.poll(Mockito.any(Duration.class))) + .thenReturn(new ConsumerRecords<>(recordMap)); Review Comment: ## CodeQL / Deprecated method or constructor invocation Invoking [ConsumerRecords.ConsumerRecords](1) should be avoided because it has been deprecated. [Show more details](https://github.com/apache/druid/security/code-scanning/11187) ########## indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/common/AcknowledgingRecordSupplier.java: ########## @@ -0,0 +1,99 @@ +/* + * 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.seekablestream.common; + +import org.apache.druid.data.input.impl.ByteEntity; + +import javax.validation.constraints.NotNull; +import java.io.Closeable; +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +/** + * Record supplier for queue-semantics streams (e.g. Kafka share groups, + * KIP-932) where the broker owns delivery state and consumers acknowledge + * records explicitly. Unlike {@link RecordSupplier}, callers do not assign + * or seek partitions; they subscribe to topics and ack individual records. + * + * @param <PartitionIdType> partition identifier type + * @param <SequenceOffsetType> sequence/offset number type + * @param <RecordType> record entity type + */ +public interface AcknowledgingRecordSupplier<PartitionIdType, SequenceOffsetType, RecordType extends ByteEntity> + extends Closeable +{ + void subscribe(Set<String> topics); + + void unsubscribe(); + + Set<String> subscription(); + + /** + * Poll for records. Records carry acquisition locks and must be + * acknowledged before the lock expires. + */ + @NotNull + List<OrderedPartitionableRecord<PartitionIdType, SequenceOffsetType, RecordType>> poll(long timeoutMs); + + /** Acknowledge a single record with the default type ({@code ACCEPT}). */ + void acknowledge(PartitionIdType partitionId, SequenceOffsetType offset); + + /** Acknowledge a single record with the given type. */ + void acknowledge(PartitionIdType partitionId, SequenceOffsetType offset, AcknowledgeType type); + + /** Acknowledge a batch of records with the given type. */ + void acknowledge(Map<PartitionIdType, Collection<SequenceOffsetType>> offsets, AcknowledgeType type); + + /** + * Commit pending acknowledgements; returns a per-partition exception when + * the commit failed for that partition. + */ + Map<PartitionIdType, Optional<Exception>> commitSync(); + + Set<PartitionIdType> getPartitionIds(String stream); Review Comment: ## CodeQL / Useless parameter The parameter 'stream' is never used. [Show more details](https://github.com/apache/druid/security/code-scanning/11188) ########## extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaShareGroupRecordSupplierTest.java: ########## @@ -0,0 +1,300 @@ +/* + * 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 org.apache.druid.data.input.kafka.KafkaRecordEntity; +import org.apache.druid.data.input.kafka.KafkaTopicPartition; +import org.apache.druid.indexing.seekablestream.common.AcknowledgeType; +import org.apache.druid.indexing.seekablestream.common.OrderedPartitionableRecord; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaShareConsumer; +import org.apache.kafka.common.KafkaException; +import org.apache.kafka.common.TopicIdPartition; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.Uuid; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mockito; + +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +public class KafkaShareGroupRecordSupplierTest +{ + private KafkaShareConsumer<byte[], byte[]> mockConsumer; + private KafkaShareGroupRecordSupplier supplier; + + @SuppressWarnings("unchecked") + @Before + public void setUp() + { + mockConsumer = Mockito.mock(KafkaShareConsumer.class); + supplier = new KafkaShareGroupRecordSupplier(mockConsumer); + } + + @After + public void tearDown() + { + supplier.close(); + } + + @Test + public void testSubscribeAndSubscription() + { + final Set<String> topics = Set.of("topic-a", "topic-b"); + Mockito.when(mockConsumer.subscription()).thenReturn(topics); + + supplier.subscribe(topics); + Mockito.verify(mockConsumer).subscribe(topics); + Assert.assertEquals(topics, supplier.subscription()); + } + + @Test + public void testUnsubscribe() + { + supplier.unsubscribe(); + Mockito.verify(mockConsumer).unsubscribe(); + } + + @Test + public void testPollWrapsRecords() + { + final ConsumerRecord<byte[], byte[]> record1 = new ConsumerRecord<>( + "test-topic", 0, 100L, "key1".getBytes(StandardCharsets.UTF_8), "value1".getBytes(StandardCharsets.UTF_8) + ); + final ConsumerRecord<byte[], byte[]> record2 = new ConsumerRecord<>( + "test-topic", 1, 200L, "key2".getBytes(StandardCharsets.UTF_8), "value2".getBytes(StandardCharsets.UTF_8) + ); + + final Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> recordMap = new HashMap<>(); + recordMap.put(new TopicPartition("test-topic", 0), List.of(record1)); + recordMap.put(new TopicPartition("test-topic", 1), List.of(record2)); + final ConsumerRecords<byte[], byte[]> consumerRecords = new ConsumerRecords<>(recordMap); + + Mockito.when(mockConsumer.poll(Mockito.any(Duration.class))).thenReturn(consumerRecords); + + final List<OrderedPartitionableRecord<KafkaTopicPartition, Long, KafkaRecordEntity>> result = + supplier.poll(1000); + + Assert.assertEquals(2, result.size()); + + final OrderedPartitionableRecord<KafkaTopicPartition, Long, KafkaRecordEntity> polled1 = + result.stream().filter(r -> r.getSequenceNumber() == 100L).findFirst().orElse(null); + Assert.assertNotNull(polled1); + Assert.assertEquals("test-topic", polled1.getStream()); + Assert.assertEquals(0, polled1.getPartitionId().partition()); + Assert.assertNotNull(polled1.getData()); + Assert.assertEquals(1, polled1.getData().size()); + } + + @Test + public void testPollReturnsEmptyOnTimeout() + { + Mockito.when(mockConsumer.poll(Mockito.any(Duration.class))).thenReturn(ConsumerRecords.empty()); + final List<OrderedPartitionableRecord<KafkaTopicPartition, Long, KafkaRecordEntity>> result = + supplier.poll(100); + Assert.assertTrue(result.isEmpty()); + } + + @Test + public void testAcknowledgeDefaultAccept() + { + final ConsumerRecord<byte[], byte[]> record = new ConsumerRecord<>( + "test-topic", 0, 42L, "key".getBytes(StandardCharsets.UTF_8), "value".getBytes(StandardCharsets.UTF_8) + ); + pollSingleRecord(record); + + final KafkaTopicPartition partition = new KafkaTopicPartition(true, "test-topic", 0); + supplier.acknowledge(partition, 42L); + + Mockito.verify(mockConsumer).acknowledge( + Mockito.same(record), + Mockito.eq(org.apache.kafka.clients.consumer.AcknowledgeType.ACCEPT) + ); + } + + @Test + public void testAcknowledgeWithRelease() + { + final ConsumerRecord<byte[], byte[]> record = new ConsumerRecord<>( + "test-topic", 0, 10L, "key".getBytes(StandardCharsets.UTF_8), "value".getBytes(StandardCharsets.UTF_8) + ); + pollSingleRecord(record); + + final KafkaTopicPartition partition = new KafkaTopicPartition(true, "test-topic", 0); + supplier.acknowledge(partition, 10L, AcknowledgeType.RELEASE); + + Mockito.verify(mockConsumer).acknowledge( + Mockito.same(record), + Mockito.eq(org.apache.kafka.clients.consumer.AcknowledgeType.RELEASE) + ); + } + + @Test + public void testAcknowledgeWithReject() + { + final ConsumerRecord<byte[], byte[]> record = new ConsumerRecord<>( + "test-topic", 0, 10L, "key".getBytes(StandardCharsets.UTF_8), "value".getBytes(StandardCharsets.UTF_8) + ); + pollSingleRecord(record); + + final KafkaTopicPartition partition = new KafkaTopicPartition(true, "test-topic", 0); + supplier.acknowledge(partition, 10L, AcknowledgeType.REJECT); + + Mockito.verify(mockConsumer).acknowledge( + Mockito.same(record), + Mockito.eq(org.apache.kafka.clients.consumer.AcknowledgeType.REJECT) + ); + } + + @Test + public void testAcknowledgeWithRenew() + { + final ConsumerRecord<byte[], byte[]> record = new ConsumerRecord<>( + "test-topic", 0, 99L, "key".getBytes(StandardCharsets.UTF_8), "value".getBytes(StandardCharsets.UTF_8) + ); + pollSingleRecord(record); + + final KafkaTopicPartition partition = new KafkaTopicPartition(true, "test-topic", 0); + supplier.acknowledge(partition, 99L, AcknowledgeType.RENEW); + + Mockito.verify(mockConsumer).acknowledge( + Mockito.same(record), + Mockito.eq(org.apache.kafka.clients.consumer.AcknowledgeType.RENEW) + ); + } + + @Test + public void testAcknowledgeBatch() + { + final List<ConsumerRecord<byte[], byte[]>> recordsP0 = Arrays.asList( + new ConsumerRecord<>("test-topic", 0, 1L, "k".getBytes(StandardCharsets.UTF_8), "v".getBytes(StandardCharsets.UTF_8)), + new ConsumerRecord<>("test-topic", 0, 2L, "k".getBytes(StandardCharsets.UTF_8), "v".getBytes(StandardCharsets.UTF_8)), + new ConsumerRecord<>("test-topic", 0, 3L, "k".getBytes(StandardCharsets.UTF_8), "v".getBytes(StandardCharsets.UTF_8)) + ); + final List<ConsumerRecord<byte[], byte[]>> recordsP1 = Arrays.asList( + new ConsumerRecord<>("test-topic", 1, 10L, "k".getBytes(StandardCharsets.UTF_8), "v".getBytes(StandardCharsets.UTF_8)), + new ConsumerRecord<>("test-topic", 1, 11L, "k".getBytes(StandardCharsets.UTF_8), "v".getBytes(StandardCharsets.UTF_8)) + ); + final Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> recordMap = new HashMap<>(); + recordMap.put(new TopicPartition("test-topic", 0), recordsP0); + recordMap.put(new TopicPartition("test-topic", 1), recordsP1); + Mockito.when(mockConsumer.poll(Mockito.any(Duration.class))) + .thenReturn(new ConsumerRecords<>(recordMap)); Review Comment: ## CodeQL / Deprecated method or constructor invocation Invoking [ConsumerRecords.ConsumerRecords](1) should be avoided because it has been deprecated. [Show more details](https://github.com/apache/druid/security/code-scanning/11186) -- 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]
