Package com.linkedin.venice.utils

Class TestUtils

java.lang.Object

com.linkedin.venice.utils.TestUtils

public class TestUtils
extends Object

General-purpose utility functions for tests.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static interface	TestUtils.NonDeterministicAssertion

Field Summary

Fields

Modifier and Type	Field	Description
static final PubSubContext	DEFAULT_PUBSUB_CONTEXT_FOR_UNIT_TESTING	**FOR UNIT TESTING PURPOSES ONLY** - Do not use in production code.

Constructor Summary

Constructors

Constructor	Description
TestUtils()

Method Summary

Modifier and Type	Method	Description
static <T extends ControllerResponse> T	assertCommand(T response)
static <T extends ControllerResponse> T	assertCommand(T response, String assertionErrorMessage)
static <T extends ControllerResponse> T	assertCommandFailure(T response, String assertionErrorMessage)
static void	checkMissingFieldInAvroRecord(org.apache.avro.generic.GenericRecord record, String fieldName)
static void	createAndVerifyStoreInAllRegions(String storeName, ControllerClient parentControllerClient, List<ControllerClient> controllerClientList)
static ETLStoreConfig	createTestETLStoreConfig()
static HybridStoreConfig	createTestHybridStoreConfig(Random random)
static PartitionerConfig	createTestPartitionerConfig(Random random)
static Store	createTestStore(String name, String owner, long createdTime)
static SystemStoreAttributes	createTestSystemStoreAttributes(String storeName, Random random)
static Map<String,SystemStoreAttributes>	createTestSystemStores(String storeName, Random random)
static List<Version>	createTestVersions(String storeName, Random random)
static Map<String,ViewConfig>	createTestViewConfigs(Random random)
static Map<String,String>	createTestViewParams(Random random)
static VersionCreationResponse	createVersionWithBatchData(ControllerClient controllerClient, String storeName, String keySchema, String valueSchema, Stream<Map.Entry> batchData, int valueSchemaId, PubSubProducerAdapterFactory pubSubProducerAdapterFactory, Map<String,String> additionalProperties, PubSubPositionTypeRegistry pubSubPositionTypeRegistry)
static VersionCreationResponse	createVersionWithBatchData(ControllerClient controllerClient, String storeName, String keySchema, String valueSchema, Stream<Map.Entry> batchData, PubSubProducerAdapterFactory pubSubProducerAdapterFactory, Map<String,String> additionalProperties, PubSubPositionTypeRegistry pubSubPositionTypeRegistry)
static void	deleteDirectory(File fileToDelete)
static String	dequalifyClassName(String className)
static boolean	directoryContainsFolder(String directoryPath, String folderName)
static List<String>	findFoldersWithFileExtension(File directory, String fileExtension)
static Map<byte[],byte[]>	generateInput(int recordCnt, boolean sorted, int startId, AvroSerializer serializer)
static String	getClusterToD2String(Map<String,String> clusterToD2)
static int	getFreePort()	WARNING: The code which generates the free port and uses it must always be called within a try/catch and a loop.
static Map<String,Object>	getIngestionIsolationPropertyMap()
static VeniceControllerMultiClusterConfig	getMultiClusterConfigFromOneCluster(VeniceControllerClusterConfig controllerConfig)
static OffsetRecord	getOffsetRecord(PubSubPosition currentPosition, Optional<PubSubPosition> endOfPushOffset, PubSubContext pubSubContext)
static SafeHelixManager	getParticipant(String cluster, String nodeId, String zkAddress, int httpPort, String stateModelDef)	Deprecated. TODO: migrate to use ServiceFactory for generating a participant
static SafeHelixManager	getParticipant(String cluster, String nodeId, String zkAddress, int httpPort, org.apache.helix.participant.statemachine.StateModelFactory<org.apache.helix.participant.statemachine.StateModel> stateModelFactory, String stateModelDef)
static Properties	getPropertiesForControllerConfig()
static Store	getRandomStore()
static StoreIngestionTaskFactory.Builder	getStoreIngestionTaskBuilder(String storeName)
static StoreIngestionTaskFactory.Builder	getStoreIngestionTaskBuilder(String storeName, boolean isHybrid)
static String	getUniqueTopicString(String prefix)
static VenicePathParser	getVenicePathParser(CompressorFactory compressorFactory, boolean decompressOnClient)
static VeniceWriterFactory	getVeniceWriterFactory(Properties properties, PubSubProducerAdapterFactory pubSubProducerAdapterFactory, PubSubPositionTypeRegistry pubSubPositionTypeRegistry)
static String	loadFileAsString(String fileName)
static Map<String,String>	mergeConfigs(List<Map<String,String>> configMaps)
static ZKStore	populateZKStore(ZKStore store, Random random)
static void	preventSystemExit()
static void	restoreSystemExit()
static List<String>	searchForFileExtension(File directory, String fileExtension)
static void	shutdownExecutor(ExecutorService executor)
static void	shutdownExecutor(ExecutorService executor, long timeout, TimeUnit unit)
static void	shutdownThread(Thread thread)
static void	shutdownThread(Thread thread, long timeout, TimeUnit unit)
static ControllerResponse	updateStoreToHybrid(String storeName, ControllerClient parentControllerClient, Optional<Boolean> enableNativeReplication, Optional<Boolean> enableActiveActiveReplication, Optional<Boolean> enableChunking)
static void	verifyDCConfigNativeAndActiveRepl(String storeName, boolean enabledNR, boolean enabledAA, ControllerClient... controllerClients)
static void	waitForNonDeterministicAssertion(long timeout, TimeUnit timeoutUnit, boolean exponentialBackOff, boolean retryOnThrowable, TestUtils.NonDeterministicAssertion assertion)
static void	waitForNonDeterministicAssertion(long timeout, TimeUnit timeoutUnit, boolean exponentialBackOff, TestUtils.NonDeterministicAssertion assertion)
static void	waitForNonDeterministicAssertion(long timeout, TimeUnit timeoutUnit, TestUtils.NonDeterministicAssertion assertion)	To be used for tests when we need to wait for an async operation to complete.
static void	waitForNonDeterministicCompletion(long timeout, TimeUnit timeoutUnit, BooleanSupplier condition)	To be used for tests when we need to wait for an async operation to complete.
static void	waitForNonDeterministicPushCompletion(String topicName, ControllerClient controllerClient, long timeout, TimeUnit timeoutUnit)	Wait for the push job for a store version or topic to be completed.
static void	writeBatchData(VersionCreationResponse response, String keySchema, String valueSchema, Stream<Map.Entry> batchData, int valueSchemaId, CompressionStrategy compressionStrategy, Function<String,ByteBuffer> compressionDictionaryGenerator, PubSubProducerAdapterFactory pubSubProducerAdapterFactory, Map<String,String> additionalProperties, PubSubPositionTypeRegistry pubSubPositionTypeRegistry)
static void	writeBatchData(VersionCreationResponse response, String keySchema, String valueSchema, Stream<Map.Entry> batchData, int valueSchemaId, PubSubProducerAdapterFactory pubSubProducerAdapterFactory, Map<String,String> additionalProperties, PubSubPositionTypeRegistry pubSubPositionTypeRegistry)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Details

DEFAULT_PUBSUB_CONTEXT_FOR_UNIT_TESTING

public static final PubSubContext DEFAULT_PUBSUB_CONTEXT_FOR_UNIT_TESTING

**FOR UNIT TESTING PURPOSES ONLY** - Do not use in production code. A pre-configured PubSubContext instance with default test values for use in unit tests. This instance is initialized with basic default components suitable for testing scenarios where a fully configured PubSubContext is not required. Production code should propagate properly configured PubSubContext instances with appropriate TopicManagerRepository and other production-ready components.

Constructor Details

TestUtils

public TestUtils()

Method Details

dequalifyClassName

public static String dequalifyClassName(String className)

waitForNonDeterministicCompletion

public static void waitForNonDeterministicCompletion(long timeout,
 TimeUnit timeoutUnit,
 BooleanSupplier condition)
                                              throws AssertionError

To be used for tests when we need to wait for an async operation to complete. Pass a timeout, and a lambda for checking if the operation is complete.

Parameters:

timeout - amount of time to wait

timeoutUnit - TimeUnit for the

condition - A BooleanSupplier which should execute the non-deterministic action and return true if it is successful, false otherwise.

Throws:

AssertionError

updateStoreToHybrid

public static ControllerResponse updateStoreToHybrid(String storeName,
 ControllerClient parentControllerClient,
 Optional<Boolean> enableNativeReplication,
 Optional<Boolean> enableActiveActiveReplication,
 Optional<Boolean> enableChunking)

waitForNonDeterministicAssertion

public static void waitForNonDeterministicAssertion(long timeout,
 TimeUnit timeoutUnit,
 TestUtils.NonDeterministicAssertion assertion)
                                             throws AssertionError

To be used for tests when we need to wait for an async operation to complete. Pass a timeout, and a labmda for checking if the operation is complete. There is an issue within Mockito where it emits VerifyError instead of ArgumentsAreDifferent Exception. Check out "ExceptionFactory#JunitArgsAreDifferent" for details. The workaround here is to catch both assert and verify error. TODO: find a better way resolve it

Parameters:

timeout - amount of time to wait

timeoutUnit - TimeUnit for the

assertion - A TestUtils.NonDeterministicAssertion which should simply execute without exception if it is successful, or throw an AssertionError otherwise.

Throws:

AssertionError - throws the exception thrown by the TestUtils.NonDeterministicAssertion if the maximum wait time has been exceeded.

waitForNonDeterministicAssertion

public static void waitForNonDeterministicAssertion(long timeout,
 TimeUnit timeoutUnit,
 boolean exponentialBackOff,
 TestUtils.NonDeterministicAssertion assertion)
                                             throws AssertionError

Throws:

AssertionError

waitForNonDeterministicAssertion

public static void waitForNonDeterministicAssertion(long timeout,
 TimeUnit timeoutUnit,
 boolean exponentialBackOff,
 boolean retryOnThrowable,
 TestUtils.NonDeterministicAssertion assertion)
                                             throws AssertionError

Throws:

AssertionError

createVersionWithBatchData

public static VersionCreationResponse createVersionWithBatchData(ControllerClient controllerClient,
 String storeName,
 String keySchema,
 String valueSchema,
 Stream<Map.Entry> batchData,
 PubSubProducerAdapterFactory pubSubProducerAdapterFactory,
 Map<String,String> additionalProperties,
 PubSubPositionTypeRegistry pubSubPositionTypeRegistry)

createVersionWithBatchData

public static VersionCreationResponse createVersionWithBatchData(ControllerClient controllerClient,
 String storeName,
 String keySchema,
 String valueSchema,
 Stream<Map.Entry> batchData,
 int valueSchemaId,
 PubSubProducerAdapterFactory pubSubProducerAdapterFactory,
 Map<String,String> additionalProperties,
 PubSubPositionTypeRegistry pubSubPositionTypeRegistry)

writeBatchData

public static void writeBatchData(VersionCreationResponse response,
 String keySchema,
 String valueSchema,
 Stream<Map.Entry> batchData,
 int valueSchemaId,
 PubSubProducerAdapterFactory pubSubProducerAdapterFactory,
 Map<String,String> additionalProperties,
 PubSubPositionTypeRegistry pubSubPositionTypeRegistry)

writeBatchData

public static void writeBatchData(VersionCreationResponse response,
 String keySchema,
 String valueSchema,
 Stream<Map.Entry> batchData,
 int valueSchemaId,
 CompressionStrategy compressionStrategy,
 Function<String,ByteBuffer> compressionDictionaryGenerator,
 PubSubProducerAdapterFactory pubSubProducerAdapterFactory,
 Map<String,String> additionalProperties,
 PubSubPositionTypeRegistry pubSubPositionTypeRegistry)

waitForNonDeterministicPushCompletion

public static void waitForNonDeterministicPushCompletion(String topicName,
 ControllerClient controllerClient,
 long timeout,
 TimeUnit timeoutUnit)

Wait for the push job for a store version or topic to be completed. The polling will fast fail if the push is found to be in ERROR state.

createTestStore

public static Store createTestStore(String name,
 String owner,
 long createdTime)

populateZKStore

public static ZKStore populateZKStore(ZKStore store,
 Random random)

createTestHybridStoreConfig

public static HybridStoreConfig createTestHybridStoreConfig(Random random)

createTestViewConfigs

public static Map<String,ViewConfig> createTestViewConfigs(Random random)

createTestViewParams

public static Map<String,String> createTestViewParams(Random random)

createTestETLStoreConfig

public static ETLStoreConfig createTestETLStoreConfig()

createTestPartitionerConfig

public static PartitionerConfig createTestPartitionerConfig(Random random)

createTestVersions

public static List<Version> createTestVersions(String storeName,
 Random random)

createTestSystemStores

public static Map<String,SystemStoreAttributes> createTestSystemStores(String storeName,
 Random random)

createTestSystemStoreAttributes

public static SystemStoreAttributes createTestSystemStoreAttributes(String storeName,
 Random random)

getParticipant

@Deprecated
public static SafeHelixManager getParticipant(String cluster,
 String nodeId,
 String zkAddress,
 int httpPort,
 String stateModelDef)

Deprecated.

TODO: migrate to use ServiceFactory for generating a participant

getParticipant

public static SafeHelixManager getParticipant(String cluster,
 String nodeId,
 String zkAddress,
 int httpPort,
 org.apache.helix.participant.statemachine.StateModelFactory<org.apache.helix.participant.statemachine.StateModel> stateModelFactory,
 String stateModelDef)

getOffsetRecord

public static OffsetRecord getOffsetRecord(PubSubPosition currentPosition,
 Optional<PubSubPosition> endOfPushOffset,
 PubSubContext pubSubContext)

getMultiClusterConfigFromOneCluster

public static VeniceControllerMultiClusterConfig getMultiClusterConfigFromOneCluster(VeniceControllerClusterConfig controllerConfig)

getPropertiesForControllerConfig

public static Properties getPropertiesForControllerConfig()

getClusterToD2String

public static String getClusterToD2String(Map<String,String> clusterToD2)

getVeniceWriterFactory

public static VeniceWriterFactory getVeniceWriterFactory(Properties properties,
 PubSubProducerAdapterFactory pubSubProducerAdapterFactory,
 PubSubPositionTypeRegistry pubSubPositionTypeRegistry)

getRandomStore

public static Store getRandomStore()

assertCommand

public static <T extends ControllerResponse> T assertCommand(T response)

assertCommand

public static <T extends ControllerResponse> T assertCommand(T response,
 String assertionErrorMessage)

assertCommandFailure

public static <T extends ControllerResponse> T assertCommandFailure(T response,
 String assertionErrorMessage)

preventSystemExit

public static void preventSystemExit()

restoreSystemExit

public static void restoreSystemExit()

createAndVerifyStoreInAllRegions

public static void createAndVerifyStoreInAllRegions(String storeName,
 ControllerClient parentControllerClient,
 List<ControllerClient> controllerClientList)

verifyDCConfigNativeAndActiveRepl

public static void verifyDCConfigNativeAndActiveRepl(String storeName,
 boolean enabledNR,
 boolean enabledAA,
 ControllerClient... controllerClients)

getStoreIngestionTaskBuilder

public static StoreIngestionTaskFactory.Builder getStoreIngestionTaskBuilder(String storeName)

getStoreIngestionTaskBuilder

public static StoreIngestionTaskFactory.Builder getStoreIngestionTaskBuilder(String storeName,
 boolean isHybrid)

generateInput

public static Map<byte[],byte[]> generateInput(int recordCnt,
 boolean sorted,
 int startId,
 AvroSerializer serializer)

shutdownThread

public static void shutdownThread(Thread thread)
                           throws InterruptedException

Throws:

InterruptedException

shutdownThread

public static void shutdownThread(Thread thread,
 long timeout,
 TimeUnit unit)
                           throws InterruptedException

Throws:

InterruptedException

shutdownExecutor

public static void shutdownExecutor(ExecutorService executor)
                             throws InterruptedException

Throws:

InterruptedException

shutdownExecutor

public static void shutdownExecutor(ExecutorService executor,
 long timeout,
 TimeUnit unit)
                             throws InterruptedException

Throws:

InterruptedException

getIngestionIsolationPropertyMap

public static Map<String,Object> getIngestionIsolationPropertyMap()

getUniqueTopicString

public static String getUniqueTopicString(String prefix)

getFreePort

public static int getFreePort()

WARNING: The code which generates the free port and uses it must always be called within a try/catch and a loop. There is no guarantee that the port returned will still be available at the time it is used. This is best-effort only.

Returns:

a free port to be used by tests.

mergeConfigs

public static Map<String,String> mergeConfigs(List<Map<String,String>> configMaps)

checkMissingFieldInAvroRecord

public static void checkMissingFieldInAvroRecord(org.apache.avro.generic.GenericRecord record,
 String fieldName)

findFoldersWithFileExtension

public static List<String> findFoldersWithFileExtension(File directory,
 String fileExtension)

searchForFileExtension

public static List<String> searchForFileExtension(File directory,
 String fileExtension)

directoryContainsFolder

public static boolean directoryContainsFolder(String directoryPath,
 String folderName)

deleteDirectory

public static void deleteDirectory(File fileToDelete)

loadFileAsString

public static String loadFileAsString(String fileName)

getVenicePathParser

public static VenicePathParser getVenicePathParser(CompressorFactory compressorFactory,
 boolean decompressOnClient)