From ddc970d7f5121968955f9a9bae886403c1bdcc7e Mon Sep 17 00:00:00 2001 From: Sushant Mane Date: Wed, 18 Dec 2024 15:08:18 -0800 Subject: [PATCH 1/3] - Updated VeniceVsonFileIterator to fail loudly on file handle acquisition failures and ensure fileReader is properly initialized. - Improved error messages and logging to include detailed information and stack traces for easier debugging. - Added null check for VeniceVsonRecordReader in the constructor. - Added VeniceVsonFileIteratorTest to test initialization and handle exceptions. --- .../vson/VeniceVsonFileIterator.java | 30 ++++-- .../vson/VeniceVsonFileIteratorTest.java | 101 ++++++++++++++++++ .../input/hdfs/TestSparkInputFromHdfs.java | 8 +- 3 files changed, 128 insertions(+), 11 deletions(-) create mode 100644 clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/recordreader/vson/VeniceVsonFileIteratorTest.java diff --git a/clients/venice-push-job/src/main/java/com/linkedin/venice/hadoop/input/recordreader/vson/VeniceVsonFileIterator.java b/clients/venice-push-job/src/main/java/com/linkedin/venice/hadoop/input/recordreader/vson/VeniceVsonFileIterator.java index 97873f0a01b..b1d386b9168 100644 --- a/clients/venice-push-job/src/main/java/com/linkedin/venice/hadoop/input/recordreader/vson/VeniceVsonFileIterator.java +++ b/clients/venice-push-job/src/main/java/com/linkedin/venice/hadoop/input/recordreader/vson/VeniceVsonFileIterator.java @@ -22,15 +22,27 @@ public class VeniceVsonFileIterator implements VeniceRecordIterator { private final BytesWritable currentKey = new BytesWritable(); private final BytesWritable currentValue = new BytesWritable(); - public VeniceVsonFileIterator(FileSystem fs, Path hdfsPath, VeniceVsonRecordReader recordReader) { - if (fs != null && hdfsPath != null) { - try { - fileReader = new SequenceFile.Reader(fs, hdfsPath, new Configuration()); - } catch (IOException e) { - LOGGER.info("Path: {} is not a sequence file.", hdfsPath.getName()); - } - } else { - throw new VeniceException("Invalid file system or path"); + public VeniceVsonFileIterator(FileSystem fileSystem, Path hdfsPath, VeniceVsonRecordReader recordReader) { + if (fileSystem == null) { + LOGGER.error("FileSystem cannot be null for VeniceVsonFileIterator"); + throw new VeniceException("FileSystem cannot be null for VeniceVsonFileIterator"); + } + if (hdfsPath == null) { + LOGGER.error("Path cannot be null for VeniceVsonFileIterator"); + throw new VeniceException("Path cannot be null for VeniceVsonFileIterator"); + } + if (recordReader == null) { + LOGGER.error("RecordReader cannot be null for VeniceVsonFileIterator"); + throw new VeniceException("RecordReader cannot be null for VeniceVsonFileIterator"); + } + + try { + this.fileReader = new SequenceFile.Reader(fileSystem, hdfsPath, new Configuration()); + } catch (IOException e) { + String errorMessage = + String.format("Failed to open file: %s. Ensure that the file is a valid sequence file.", hdfsPath.getName()); + LOGGER.error(errorMessage, e); + throw new VeniceException(errorMessage, e); } this.recordReader = recordReader; diff --git a/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/recordreader/vson/VeniceVsonFileIteratorTest.java b/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/recordreader/vson/VeniceVsonFileIteratorTest.java new file mode 100644 index 00000000000..e07cfff4f07 --- /dev/null +++ b/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/recordreader/vson/VeniceVsonFileIteratorTest.java @@ -0,0 +1,101 @@ +package com.linkedin.venice.hadoop.input.recordreader.vson; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.when; +import static org.testng.Assert.*; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.expectThrows; + +import com.linkedin.venice.exceptions.VeniceException; +import com.linkedin.venice.spark.input.hdfs.TestSparkInputFromHdfs; +import com.linkedin.venice.utils.Utils; +import java.io.File; +import java.io.IOException; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.mockito.Mockito; +import org.testng.annotations.Test; + + +public class VeniceVsonFileIteratorTest { + @Test + public void testConstructorInitializationAndExceptions() throws IOException { + // Setup: Temporary directory and files + File tempDir = Utils.getTempDataDirectory("test-dir"); + File tempSeqFile = File.createTempFile("test-file", ".seq", tempDir); + Path tempSeqFilePath = new Path(tempSeqFile.getAbsolutePath()); + File tempNonSeqFile = File.createTempFile("test-file", ".txt", tempDir); + Path tempNonSeqFilePath = new Path(tempNonSeqFile.getAbsolutePath()); + + FileSystem fileSystem = FileSystem.getLocal(new Configuration()); + VeniceVsonRecordReader mockRecordReader = Mockito.mock(VeniceVsonRecordReader.class); + + // Write valid data to the sequence file + TestSparkInputFromHdfs.writeVsonFile(tempSeqFilePath, 1, 2); + + // Case 1: Valid FileSystem and Path + VeniceVsonFileIterator iterator = new VeniceVsonFileIterator(fileSystem, tempSeqFilePath, mockRecordReader); + assertNotNull(iterator); + + // Case 2: Null FileSystem + Exception exception = + expectThrows(VeniceException.class, () -> new VeniceVsonFileIterator(null, tempSeqFilePath, mockRecordReader)); + assertTrue(exception.getMessage().contains("FileSystem cannot be null")); + + // Case 3: Null Path + Exception exception2 = + expectThrows(VeniceException.class, () -> new VeniceVsonFileIterator(fileSystem, null, mockRecordReader)); + assertTrue(exception2.getMessage().contains("Path cannot be null")); + + // Case 4: Null RecordReader + Exception exception3 = + expectThrows(VeniceException.class, () -> new VeniceVsonFileIterator(fileSystem, tempSeqFilePath, null)); + assertTrue(exception3.getMessage().contains("RecordReader cannot be null")); + + // Case 4: Invalid SequenceFile + expectThrows( + VeniceException.class, + () -> new VeniceVsonFileIterator(fileSystem, tempNonSeqFilePath, mockRecordReader)); + } + + @Test + public void testNextAndGetCurrentKeyValue() throws IOException { + // Setup: Temporary sequence file + File tempDir = Utils.getTempDataDirectory("test-dir"); + File tempSeqFile = File.createTempFile("test-file", ".seq", tempDir); + Path tempSeqFilePath = new Path(tempSeqFile.getAbsolutePath()); + FileSystem fileSystem = FileSystem.getLocal(new Configuration()); + VeniceVsonRecordReader mockRecordReader = Mockito.mock(VeniceVsonRecordReader.class); + + int totalRecords = 5; + // Write valid data to the sequence file + TestSparkInputFromHdfs.writeVsonFile(tempSeqFilePath, 1, totalRecords); + + VeniceVsonFileIterator vsonFileIterator = new VeniceVsonFileIterator(fileSystem, tempSeqFilePath, mockRecordReader); + + // Mock behavior for record reader + when(mockRecordReader.getKeyBytes(any(), any())).thenReturn("mockKey".getBytes()); + when(mockRecordReader.getValueBytes(any(), any())).thenReturn("mockValue".getBytes()); + + byte[] lastKey = null; + byte[] lastValue = null; + + int expectedRecords = totalRecords; + while (vsonFileIterator.next()) { + lastKey = vsonFileIterator.getCurrentKey(); + lastValue = vsonFileIterator.getCurrentValue(); + assertNotNull(lastKey); + assertNotNull(lastValue); + expectedRecords -= 1; + } + assertEquals(expectedRecords, 0); + + // Case 2: No more records (simulate by mocking behavior) + assertFalse(vsonFileIterator.next()); + assertNotNull(lastKey); + assertNotNull(lastValue); + assertEquals(lastKey, vsonFileIterator.getCurrentKey()); + assertEquals(lastValue, vsonFileIterator.getCurrentValue()); + } +} diff --git a/clients/venice-push-job/src/test/java/com/linkedin/venice/spark/input/hdfs/TestSparkInputFromHdfs.java b/clients/venice-push-job/src/test/java/com/linkedin/venice/spark/input/hdfs/TestSparkInputFromHdfs.java index 927bb290577..11a2c63ae99 100644 --- a/clients/venice-push-job/src/test/java/com/linkedin/venice/spark/input/hdfs/TestSparkInputFromHdfs.java +++ b/clients/venice-push-job/src/test/java/com/linkedin/venice/spark/input/hdfs/TestSparkInputFromHdfs.java @@ -272,7 +272,11 @@ private void writeAvroFile(File inputDir, String fileName, int start, int end) t } } - private void writeVsonFile(File parentDir, String fileName, int start, int end) throws IOException { + public static void writeVsonFile(File parentDir, String fileName, int start, int end) throws IOException { + writeVsonFile(new Path(parentDir.getAbsolutePath(), fileName), start, end); + } + + public static void writeVsonFile(Path path, int start, int end) throws IOException { SequenceFile.Metadata metadata = new SequenceFile.Metadata(); metadata.set(new Text("key.schema"), new Text(VSON_STRING_SCHEMA)); metadata.set(new Text("value.schema"), new Text(VSON_STRING_SCHEMA)); @@ -281,7 +285,7 @@ private void writeVsonFile(File parentDir, String fileName, int start, int end) try (SequenceFile.Writer writer = SequenceFile.createWriter( new Configuration(), - SequenceFile.Writer.file(new Path(parentDir.toString(), fileName)), + SequenceFile.Writer.file(path), SequenceFile.Writer.keyClass(BytesWritable.class), SequenceFile.Writer.valueClass(BytesWritable.class), SequenceFile.Writer.metadata(metadata))) { From 6d0c4bc0658b4dc823acde3b0fecfdcb87c45865 Mon Sep 17 00:00:00 2001 From: Sushant Mane Date: Wed, 18 Dec 2024 16:09:45 -0800 Subject: [PATCH 2/3] Replace with single imports --- .../davinci/stats/DIVStatsReporterTest.java | 5 +++-- .../store/memory/InMemoryStoragePartitionTest.java | 2 -- .../java/com/linkedin/davinci/utils/MapTest.java | 4 +++- .../transport/GrpcTransportClientTest.java | 14 ++++++++++++-- .../input/kafka/TestKafkaInputKeyComparator.java | 2 +- .../input/kafka/TestKafkaInputMRPartitioner.java | 3 ++- .../TestKafkaInputValueGroupingComparator.java | 2 +- .../vson/VeniceVsonFileIteratorTest.java | 4 +++- .../pulsar/sink/VenicePulsarSinkConfigTest.java | 4 +++- .../RouterBasedHybridStoreQuotaMonitorTest.java | 2 -- .../linkedin/venice/compute/ComputeUtilsTest.java | 7 +++++-- .../linkedin/venice/meta/TestStoreJsonSerde.java | 2 +- .../kafka/admin/ApacheKafkaAdminConfigTest.java | 2 +- .../linkedin/venice/utils/KafkaSSLUtilsTest.java | 3 ++- .../meta/RequestBasedMetadataIntegrationTest.java | 5 ++++- 15 files changed, 41 insertions(+), 20 deletions(-) diff --git a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/stats/DIVStatsReporterTest.java b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/stats/DIVStatsReporterTest.java index a5279bc6a9f..9c7a54254cb 100644 --- a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/stats/DIVStatsReporterTest.java +++ b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/stats/DIVStatsReporterTest.java @@ -1,8 +1,9 @@ package com.linkedin.davinci.stats; import static com.linkedin.venice.stats.StatsErrorCode.NULL_DIV_STATS; -import static org.mockito.Mockito.*; -import static org.testng.Assert.*; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.testng.Assert.assertEquals; import com.linkedin.venice.tehuti.MockTehutiReporter; import com.linkedin.venice.utils.Utils; diff --git a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/store/memory/InMemoryStoragePartitionTest.java b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/store/memory/InMemoryStoragePartitionTest.java index 1fbcc29ca1e..dd48e740ad6 100644 --- a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/store/memory/InMemoryStoragePartitionTest.java +++ b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/store/memory/InMemoryStoragePartitionTest.java @@ -1,7 +1,5 @@ package com.linkedin.davinci.store.memory; -import static org.testng.Assert.*; - import java.nio.ByteBuffer; import org.testng.Assert; import org.testng.annotations.Test; diff --git a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/utils/MapTest.java b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/utils/MapTest.java index ca1a26a43a7..a3460bdbfad 100644 --- a/clients/da-vinci-client/src/test/java/com/linkedin/davinci/utils/MapTest.java +++ b/clients/da-vinci-client/src/test/java/com/linkedin/davinci/utils/MapTest.java @@ -1,6 +1,8 @@ package com.linkedin.davinci.utils; -import static org.testng.Assert.*; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertTrue; import java.util.ArrayList; import java.util.HashMap; diff --git a/clients/venice-client/src/test/java/com/linkedin/venice/fastclient/transport/GrpcTransportClientTest.java b/clients/venice-client/src/test/java/com/linkedin/venice/fastclient/transport/GrpcTransportClientTest.java index 02b228ce1b3..d1e3a8659ac 100644 --- a/clients/venice-client/src/test/java/com/linkedin/venice/fastclient/transport/GrpcTransportClientTest.java +++ b/clients/venice-client/src/test/java/com/linkedin/venice/fastclient/transport/GrpcTransportClientTest.java @@ -1,7 +1,17 @@ package com.linkedin.venice.fastclient.transport; -import static org.mockito.Mockito.*; -import static org.testng.Assert.*; +import static org.mockito.Mockito.any; +import static org.mockito.Mockito.anyBoolean; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertTrue; import com.google.common.collect.ImmutableMap; import com.linkedin.r2.transport.common.Client; diff --git a/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/kafka/TestKafkaInputKeyComparator.java b/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/kafka/TestKafkaInputKeyComparator.java index e85971a2c50..8db05df6eb6 100644 --- a/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/kafka/TestKafkaInputKeyComparator.java +++ b/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/kafka/TestKafkaInputKeyComparator.java @@ -1,6 +1,6 @@ package com.linkedin.venice.hadoop.input.kafka; -import static org.testng.Assert.*; +import static org.testng.Assert.assertTrue; import com.linkedin.venice.hadoop.input.kafka.avro.KafkaInputMapperKey; import com.linkedin.venice.serializer.FastSerializerDeserializerFactory; diff --git a/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/kafka/TestKafkaInputMRPartitioner.java b/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/kafka/TestKafkaInputMRPartitioner.java index dc887e8b22a..8eaaa3a783e 100644 --- a/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/kafka/TestKafkaInputMRPartitioner.java +++ b/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/kafka/TestKafkaInputMRPartitioner.java @@ -1,6 +1,7 @@ package com.linkedin.venice.hadoop.input.kafka; -import static org.testng.Assert.*; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNotEquals; import org.apache.hadoop.io.BytesWritable; import org.apache.hadoop.mapred.JobConf; diff --git a/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/kafka/TestKafkaInputValueGroupingComparator.java b/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/kafka/TestKafkaInputValueGroupingComparator.java index ba3dfc6876e..a8c4d0bd6c1 100644 --- a/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/kafka/TestKafkaInputValueGroupingComparator.java +++ b/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/kafka/TestKafkaInputValueGroupingComparator.java @@ -1,6 +1,6 @@ package com.linkedin.venice.hadoop.input.kafka; -import static org.testng.Assert.*; +import static org.testng.Assert.assertTrue; import org.apache.hadoop.io.BytesWritable; import org.testng.annotations.Test; diff --git a/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/recordreader/vson/VeniceVsonFileIteratorTest.java b/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/recordreader/vson/VeniceVsonFileIteratorTest.java index e07cfff4f07..b76478fb504 100644 --- a/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/recordreader/vson/VeniceVsonFileIteratorTest.java +++ b/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/recordreader/vson/VeniceVsonFileIteratorTest.java @@ -2,8 +2,10 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.when; -import static org.testng.Assert.*; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertTrue; import static org.testng.Assert.expectThrows; import com.linkedin.venice.exceptions.VeniceException; diff --git a/integrations/venice-pulsar/src/test/java/com/linkedin/venice/pulsar/sink/VenicePulsarSinkConfigTest.java b/integrations/venice-pulsar/src/test/java/com/linkedin/venice/pulsar/sink/VenicePulsarSinkConfigTest.java index 222aaf87940..93812748e09 100644 --- a/integrations/venice-pulsar/src/test/java/com/linkedin/venice/pulsar/sink/VenicePulsarSinkConfigTest.java +++ b/integrations/venice-pulsar/src/test/java/com/linkedin/venice/pulsar/sink/VenicePulsarSinkConfigTest.java @@ -1,6 +1,8 @@ package com.linkedin.venice.pulsar.sink; -import static org.testng.Assert.*; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertTrue; import java.util.Map; import org.testng.annotations.Test; diff --git a/integrations/venice-samza/src/test/java/com/linkedin/venice/pushmonitor/RouterBasedHybridStoreQuotaMonitorTest.java b/integrations/venice-samza/src/test/java/com/linkedin/venice/pushmonitor/RouterBasedHybridStoreQuotaMonitorTest.java index b0e0eef746a..9114084b3b5 100644 --- a/integrations/venice-samza/src/test/java/com/linkedin/venice/pushmonitor/RouterBasedHybridStoreQuotaMonitorTest.java +++ b/integrations/venice-samza/src/test/java/com/linkedin/venice/pushmonitor/RouterBasedHybridStoreQuotaMonitorTest.java @@ -1,7 +1,5 @@ package com.linkedin.venice.pushmonitor; -import static org.testng.Assert.*; - import com.fasterxml.jackson.databind.ObjectMapper; import com.linkedin.venice.client.store.transport.TransportClient; import com.linkedin.venice.client.store.transport.TransportClientResponse; diff --git a/internal/venice-client-common/src/test/java/com/linkedin/venice/compute/ComputeUtilsTest.java b/internal/venice-client-common/src/test/java/com/linkedin/venice/compute/ComputeUtilsTest.java index c38c40b98b4..f9dabf6d6ea 100644 --- a/internal/venice-client-common/src/test/java/com/linkedin/venice/compute/ComputeUtilsTest.java +++ b/internal/venice-client-common/src/test/java/com/linkedin/venice/compute/ComputeUtilsTest.java @@ -1,7 +1,10 @@ package com.linkedin.venice.compute; -import static com.linkedin.venice.utils.TestWriteUtils.*; -import static org.testng.Assert.*; +import static com.linkedin.venice.utils.TestWriteUtils.loadFileAsString; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNull; +import static org.testng.Assert.assertThrows; +import static org.testng.Assert.assertTrue; import com.linkedin.avro.api.PrimitiveFloatList; import com.linkedin.avro.fastserde.primitive.PrimitiveFloatArrayList; diff --git a/internal/venice-common/src/test/java/com/linkedin/venice/meta/TestStoreJsonSerde.java b/internal/venice-common/src/test/java/com/linkedin/venice/meta/TestStoreJsonSerde.java index 72954d690b5..4d42705285e 100644 --- a/internal/venice-common/src/test/java/com/linkedin/venice/meta/TestStoreJsonSerde.java +++ b/internal/venice-common/src/test/java/com/linkedin/venice/meta/TestStoreJsonSerde.java @@ -1,6 +1,6 @@ package com.linkedin.venice.meta; -import static org.testng.Assert.*; +import static org.testng.Assert.assertEquals; import com.linkedin.venice.ConfigConstants; import com.linkedin.venice.helix.StoreJSONSerializer; diff --git a/internal/venice-common/src/test/java/com/linkedin/venice/pubsub/adapter/kafka/admin/ApacheKafkaAdminConfigTest.java b/internal/venice-common/src/test/java/com/linkedin/venice/pubsub/adapter/kafka/admin/ApacheKafkaAdminConfigTest.java index 1f2a998d7ba..6058c5cf5ce 100644 --- a/internal/venice-common/src/test/java/com/linkedin/venice/pubsub/adapter/kafka/admin/ApacheKafkaAdminConfigTest.java +++ b/internal/venice-common/src/test/java/com/linkedin/venice/pubsub/adapter/kafka/admin/ApacheKafkaAdminConfigTest.java @@ -1,6 +1,6 @@ package com.linkedin.venice.pubsub.adapter.kafka.admin; -import static org.testng.Assert.*; +import static org.testng.Assert.assertEquals; import com.linkedin.venice.pubsub.api.PubSubSecurityProtocol; import com.linkedin.venice.utils.VeniceProperties; diff --git a/internal/venice-common/src/test/java/com/linkedin/venice/utils/KafkaSSLUtilsTest.java b/internal/venice-common/src/test/java/com/linkedin/venice/utils/KafkaSSLUtilsTest.java index ef67ef05aee..03bbfb87914 100644 --- a/internal/venice-common/src/test/java/com/linkedin/venice/utils/KafkaSSLUtilsTest.java +++ b/internal/venice-common/src/test/java/com/linkedin/venice/utils/KafkaSSLUtilsTest.java @@ -1,6 +1,7 @@ package com.linkedin.venice.utils; -import static org.testng.Assert.*; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertTrue; import com.linkedin.venice.pubsub.api.PubSubSecurityProtocol; import org.testng.annotations.Test; diff --git a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/fastclient/meta/RequestBasedMetadataIntegrationTest.java b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/fastclient/meta/RequestBasedMetadataIntegrationTest.java index 79a63e11380..bbbdfe68dd3 100644 --- a/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/fastclient/meta/RequestBasedMetadataIntegrationTest.java +++ b/internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/fastclient/meta/RequestBasedMetadataIntegrationTest.java @@ -1,7 +1,10 @@ package com.linkedin.venice.fastclient.meta; import static com.linkedin.venice.ConfigKeys.SERVER_HTTP2_INBOUND_ENABLED; -import static org.testng.Assert.*; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertTrue; import com.linkedin.d2.balancer.D2Client; import com.linkedin.r2.transport.common.Client; From 285323e044c54abf0e7c2cdee6f4eeef13563c37 Mon Sep 17 00:00:00 2001 From: Sushant Mane Date: Wed, 18 Dec 2024 19:42:50 -0800 Subject: [PATCH 3/3] Add avro test --- .../vson/VeniceVsonFileIteratorTest.java | 14 +++++++++++++- .../spark/input/hdfs/TestSparkInputFromHdfs.java | 2 +- 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/recordreader/vson/VeniceVsonFileIteratorTest.java b/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/recordreader/vson/VeniceVsonFileIteratorTest.java index b76478fb504..63664d33f57 100644 --- a/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/recordreader/vson/VeniceVsonFileIteratorTest.java +++ b/clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/recordreader/vson/VeniceVsonFileIteratorTest.java @@ -10,6 +10,7 @@ import com.linkedin.venice.exceptions.VeniceException; import com.linkedin.venice.spark.input.hdfs.TestSparkInputFromHdfs; +import com.linkedin.venice.utils.ExceptionUtils; import com.linkedin.venice.utils.Utils; import java.io.File; import java.io.IOException; @@ -56,9 +57,20 @@ public void testConstructorInitializationAndExceptions() throws IOException { assertTrue(exception3.getMessage().contains("RecordReader cannot be null")); // Case 4: Invalid SequenceFile - expectThrows( + Exception exception4 = expectThrows( VeniceException.class, () -> new VeniceVsonFileIterator(fileSystem, tempNonSeqFilePath, mockRecordReader)); + assertTrue(ExceptionUtils.recursiveMessageContains(exception4, "not a SequenceFile")); + + // Create avro file + File tempAvroFile = File.createTempFile("test-file", ".avro", tempDir); + Path tempAvroFilePath = new Path(tempAvroFile.getAbsolutePath()); + TestSparkInputFromHdfs.writeAvroFile(tempDir, tempAvroFilePath.getName(), 1, 2); + + Exception exception5 = expectThrows( + VeniceException.class, + () -> new VeniceVsonFileIterator(fileSystem, tempAvroFilePath, mockRecordReader)); + assertTrue(ExceptionUtils.recursiveMessageContains(exception5, "not a SequenceFile")); } @Test diff --git a/clients/venice-push-job/src/test/java/com/linkedin/venice/spark/input/hdfs/TestSparkInputFromHdfs.java b/clients/venice-push-job/src/test/java/com/linkedin/venice/spark/input/hdfs/TestSparkInputFromHdfs.java index 11a2c63ae99..4fe17daf9cd 100644 --- a/clients/venice-push-job/src/test/java/com/linkedin/venice/spark/input/hdfs/TestSparkInputFromHdfs.java +++ b/clients/venice-push-job/src/test/java/com/linkedin/venice/spark/input/hdfs/TestSparkInputFromHdfs.java @@ -257,7 +257,7 @@ private Map getDefaultConfigs(File inputDir) { return config; } - private void writeAvroFile(File inputDir, String fileName, int start, int end) throws IOException { + public static void writeAvroFile(File inputDir, String fileName, int start, int end) throws IOException { File file = new File(inputDir, fileName); DatumWriter datumWriter = new GenericDatumWriter<>(AVRO_FILE_SCHEMA);