-
Notifications
You must be signed in to change notification settings - Fork 85
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[controller][test] Message header triggered new KME schema registrati…
…on in child controller As part of the solution to remove deployment order for KafkaMessageEnvelope (KME) value schema, for controllers, when they find an unknown KME schema from the message header, they need to (talk to the system cluster leader controller to) register the new schema into local colo system store. Otherwise, as KME is not embedded in every Kafka message, if servers restart and resume the consumption from a non-SOS message with new KME (for example, lor1 servers could consume TS sent by lva1 controller), servers can not find the right schema to deserialize it and keep failing. This rb enables the child controllers to register unknown KME schemas when discovering them from the message's header. It maily leverages the existing functionailities from the ControllerClientBackedSystemSchemaInitializer class for the implementation. A new config is introduced to enable this feature and a new integration test is added to verify the correctness of the new feature.
- Loading branch information
Showing
13 changed files
with
312 additions
and
32 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
131 changes: 131 additions & 0 deletions
131
...t/java/com/linkedin/venice/controller/TestControllerKMERegistrationFromMessageHeader.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,131 @@ | ||
| package com.linkedin.venice.controller; | ||
|
|
||
| import static com.linkedin.venice.ConfigKeys.DEFAULT_MAX_NUMBER_OF_PARTITIONS; | ||
| import static com.linkedin.venice.ConfigKeys.DEFAULT_NUMBER_OF_PARTITION_FOR_HYBRID; | ||
| import static com.linkedin.venice.ConfigKeys.DEFAULT_PARTITION_SIZE; | ||
| import static com.linkedin.venice.ConfigKeys.KME_REGISTRATION_FROM_MESSAGE_HEADER_ENABLED; | ||
|
|
||
| import com.linkedin.venice.helix.HelixReadWriteSchemaRepository; | ||
| import com.linkedin.venice.helix.HelixReadWriteSchemaRepositoryAdapter; | ||
| import com.linkedin.venice.integration.utils.ServiceFactory; | ||
| import com.linkedin.venice.integration.utils.VeniceControllerWrapper; | ||
| import com.linkedin.venice.integration.utils.VeniceMultiClusterWrapper; | ||
| import com.linkedin.venice.integration.utils.VeniceTwoLayerMultiRegionMultiClusterWrapper; | ||
| import com.linkedin.venice.serialization.avro.AvroProtocolDefinition; | ||
| import com.linkedin.venice.serialization.avro.KafkaValueSerializer; | ||
| import com.linkedin.venice.utils.TestUtils; | ||
| import com.linkedin.venice.utils.Utils; | ||
| import java.util.List; | ||
| import java.util.Optional; | ||
| import java.util.Properties; | ||
| import java.util.concurrent.TimeUnit; | ||
| import java.util.stream.IntStream; | ||
| import org.apache.logging.log4j.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
| import org.testng.Assert; | ||
| import org.testng.annotations.AfterClass; | ||
| import org.testng.annotations.BeforeClass; | ||
| import org.testng.annotations.Test; | ||
|
|
||
|
|
||
| public class TestControllerKMERegistrationFromMessageHeader { | ||
| private static final Logger LOGGER = LogManager.getLogger(TestControllerKMERegistrationFromMessageHeader.class); | ||
| private static final int TEST_TIMEOUT = 90_000; // ms | ||
| private static final int NUMBER_OF_CHILD_DATACENTERS = 1; | ||
| private static final int NUMBER_OF_CLUSTERS = 1; | ||
|
|
||
| private static final String[] CLUSTER_NAMES = | ||
| IntStream.range(0, NUMBER_OF_CLUSTERS).mapToObj(i -> "venice-cluster" + i).toArray(String[]::new); | ||
|
|
||
| private List<VeniceMultiClusterWrapper> childDatacenters; | ||
| private VeniceTwoLayerMultiRegionMultiClusterWrapper multiRegionMultiClusterWrapper; | ||
|
|
||
| @BeforeClass | ||
| public void setUp() { | ||
| Properties controllerProps = new Properties(); | ||
| controllerProps.put(DEFAULT_NUMBER_OF_PARTITION_FOR_HYBRID, 2); | ||
| controllerProps.put(DEFAULT_MAX_NUMBER_OF_PARTITIONS, 3); | ||
| controllerProps.put(DEFAULT_PARTITION_SIZE, 1024); | ||
| controllerProps.put(KME_REGISTRATION_FROM_MESSAGE_HEADER_ENABLED, true); | ||
| multiRegionMultiClusterWrapper = ServiceFactory.getVeniceTwoLayerMultiRegionMultiClusterWrapper( | ||
| NUMBER_OF_CHILD_DATACENTERS, | ||
| NUMBER_OF_CLUSTERS, | ||
| 1, | ||
| 3, | ||
| 1, | ||
| 1, | ||
| 1, | ||
| Optional.of(controllerProps), | ||
| Optional.of(controllerProps), | ||
| Optional.empty()); | ||
|
|
||
| childDatacenters = multiRegionMultiClusterWrapper.getChildRegions(); | ||
| } | ||
|
|
||
| @AfterClass(alwaysRun = true) | ||
| public void cleanUp() { | ||
| Utils.closeQuietlyWithErrorLogged(multiRegionMultiClusterWrapper); | ||
| } | ||
|
|
||
| @Test(timeOut = TEST_TIMEOUT) | ||
| public void testKMERegistrationThroughAdminTopicChannel() { | ||
| String clusterName = CLUSTER_NAMES[0]; | ||
| String storeName = Utils.getUniqueString("store"); | ||
|
|
||
| VeniceControllerWrapper pController = | ||
| multiRegionMultiClusterWrapper.getLeaderParentControllerWithRetries(clusterName); | ||
|
|
||
| VeniceMultiClusterWrapper child = childDatacenters.get(0); | ||
|
|
||
| // Remove the latest schema from child controller's local value serializer and remove it from child colo's schema | ||
| // repository (ZK). | ||
| VeniceControllerWrapper leaderController = child.getLeaderController(clusterName); | ||
| KafkaValueSerializer valueSerializer = | ||
| leaderController.getController().getVeniceControllerService().getKafkaValueSerializer(); | ||
| TestUtils.waitForNonDeterministicAssertion(30, TimeUnit.SECONDS, true, () -> { | ||
| Assert.assertTrue( | ||
| valueSerializer.getProtocolVersionSize() >= AvroProtocolDefinition.KAFKA_MESSAGE_ENVELOPE | ||
| .getCurrentProtocolVersion()); | ||
| }); | ||
| HelixReadWriteSchemaRepositoryAdapter adapter = | ||
| (HelixReadWriteSchemaRepositoryAdapter) (leaderController.getVeniceHelixAdmin() | ||
| .getHelixVeniceClusterResources(clusterName) | ||
| .getSchemaRepository()); | ||
| HelixReadWriteSchemaRepository repo = | ||
| (HelixReadWriteSchemaRepository) adapter.getReadWriteRegularStoreSchemaRepository(); | ||
| repo.removeValueSchema( | ||
| AvroProtocolDefinition.KAFKA_MESSAGE_ENVELOPE.getSystemStoreName(), | ||
| AvroProtocolDefinition.KAFKA_MESSAGE_ENVELOPE.getCurrentProtocolVersion()); | ||
| valueSerializer.removeAllSchemas(); | ||
| LOGGER.info("all schemas are removed"); | ||
|
|
||
| // Verify that the latest version of the protocol is deleted in ZK. | ||
| TestUtils.waitForNonDeterministicAssertion(30, TimeUnit.SECONDS, true, () -> { | ||
| Assert.assertTrue( | ||
| repo.getValueSchema( | ||
| AvroProtocolDefinition.KAFKA_MESSAGE_ENVELOPE.getSystemStoreName(), | ||
| AvroProtocolDefinition.KAFKA_MESSAGE_ENVELOPE.getCurrentProtocolVersion()) == null); | ||
| }); | ||
|
|
||
| /* | ||
| * Calling parent controller's create store action which will trigger an admin message which contains | ||
| * the latest schema in its header, child controller when it encounters the new schema in the message header, | ||
| * would register the new schema into the child colo's schema repo as well as add to its local serializer. | ||
| */ | ||
|
|
||
| pController.getVeniceAdmin().createStore(clusterName, storeName, "", "\"string\"", "\"string\"", false); | ||
|
|
||
| // Verify that schema is registered in the child colo's schema repo. | ||
| TestUtils.waitForNonDeterministicAssertion(30, TimeUnit.SECONDS, true, () -> { | ||
| Assert.assertTrue( | ||
| repo.getValueSchema( | ||
| AvroProtocolDefinition.KAFKA_MESSAGE_ENVELOPE.getSystemStoreName(), | ||
| AvroProtocolDefinition.KAFKA_MESSAGE_ENVELOPE.getCurrentProtocolVersion()) != null); | ||
| }); | ||
|
|
||
| // Verify that store is created successfully in the child colo from the child controller's view. | ||
| TestUtils.waitForNonDeterministicAssertion(30, TimeUnit.SECONDS, true, () -> { | ||
| Assert.assertTrue(leaderController.getVeniceAdmin().getStore(clusterName, storeName).getName().equals(storeName)); | ||
| }); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.