Skip to content

Commit

Permalink
[controller][admin-tool] Check for deleted store in admin execution t…
Browse files Browse the repository at this point in the history 
…ask.
  • Loading branch information
Sourav Maji committed Feb 24, 2025
1 parent 2f3a731 commit ba156c7
Showing 1 changed file with 72 additions and 0 deletions.
72 changes: 72 additions & 0 deletions ...oller/src/main/java/com/linkedin/venice/controller/kafka/consumer/AdminExecutionTask.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -315,6 +315,15 @@ private void handleStoreCreation(StoreCreation message) {
private void handleValueSchemaCreation(ValueSchemaCreation message) {
String clusterName = message.clusterName.toString();
String storeName = message.storeName.toString();

if (!admin.hasStore(clusterName, storeName)) {
LOGGER.warn(
"Ignoring value schema creation command for non-existent store {} in cluster {}",
storeName,
clusterName);
return;
}

String schemaStr = message.schema.definition.toString();
final int schemaId = message.schemaId;

Expand Down Expand Up @@ -364,6 +373,15 @@ private void handleSupersetSchemaCreation(SupersetSchemaCreation message) {
private void handleDisableStoreWrite(PauseStore message) {
String clusterName = message.clusterName.toString();
String storeName = message.storeName.toString();

if (!admin.hasStore(clusterName, storeName)) {
LOGGER.warn(
"Ignoring disable store write command for non-existent store {} in cluster {}",
storeName,
clusterName);
return;
}

admin.setStoreWriteability(clusterName, storeName, false);

LOGGER.info("Disabled store to write: {} in cluster: {}", storeName, clusterName);
Expand All @@ -372,6 +390,13 @@ private void handleDisableStoreWrite(PauseStore message) {
private void handleEnableStoreWrite(ResumeStore message) {
String clusterName = message.clusterName.toString();
String storeName = message.storeName.toString();

if (!admin.hasStore(clusterName, storeName)) {
LOGGER
.warn("Ignoring enable store write command for non-existent store {} in cluster {}", storeName, clusterName);
return;
}

admin.setStoreWriteability(clusterName, storeName, true);

LOGGER.info("Enabled store to write: {} in cluster: {}", storeName, clusterName);
Expand All @@ -380,6 +405,13 @@ private void handleEnableStoreWrite(ResumeStore message) {
private void handleDisableStoreRead(DisableStoreRead message) {
String clusterName = message.clusterName.toString();
String storeName = message.storeName.toString();

if (!admin.hasStore(clusterName, storeName)) {
LOGGER
.warn("Ignoring disable store read command for non-existent store {} in cluster {}", storeName, clusterName);
return;
}

admin.setStoreReadability(clusterName, storeName, false);

LOGGER.info("Disabled store to read: {} in cluster: {}", storeName, clusterName);
Expand All @@ -388,6 +420,13 @@ private void handleDisableStoreRead(DisableStoreRead message) {
private void handleEnableStoreRead(EnableStoreRead message) {
String clusterName = message.clusterName.toString();
String storeName = message.storeName.toString();

if (!admin.hasStore(clusterName, storeName)) {
LOGGER
.warn("Ignoring enable store read command for non-existent store {} in cluster {}", storeName, clusterName);
return;
}

admin.setStoreReadability(clusterName, storeName, true);
LOGGER.info("Enabled store to read: {} in cluster: {}", storeName, clusterName);
}
Expand All @@ -407,6 +446,15 @@ private void handleKillOfflinePushJob(KillOfflinePushJob message) {
private void handleDeleteAllVersions(DeleteAllVersions message) {
String clusterName = message.clusterName.toString();
String storeName = message.storeName.toString();

if (!admin.hasStore(clusterName, storeName)) {
LOGGER.warn(
"Ignoring delete store versions command for non-existent store {} in cluster {}",
storeName,
clusterName);
return;
}

admin.deleteAllVersionsInStore(clusterName, storeName);
LOGGER.info("Deleted all of version in store: {} in cluster: {}", storeName, clusterName);
}
Expand All @@ -415,6 +463,11 @@ private void handleDeleteOldVersion(DeleteOldVersion message) {
String clusterName = message.clusterName.toString();
String storeName = message.storeName.toString();
int versionNum = message.versionNum;
if (!admin.hasStore(clusterName, storeName)) {
LOGGER
.warn("Ignoring delete-old-version command for non-existent store {} in cluster {}", storeName, clusterName);
return;
}
// Delete an old version for a Venice store.
admin.deleteOldVersionInStore(clusterName, storeName, versionNum);
LOGGER.info("Deleted version: {} in store: {} in cluster: {}", versionNum, storeName, clusterName);
Expand All @@ -424,6 +477,13 @@ private void handleSetStoreCurrentVersion(SetStoreCurrentVersion message) {
String clusterName = message.clusterName.toString();
String storeName = message.storeName.toString();
int version = message.currentVersion;

if (!admin.hasStore(clusterName, storeName)) {
LOGGER
.warn("Ignoring set-current-version command for non-existent store {} in cluster {}", storeName, clusterName);
return;
}

admin.setStoreCurrentVersion(clusterName, storeName, version);

LOGGER.info("Set store: {} version to {} in cluster: {}", storeName, version, clusterName);
Expand All @@ -442,6 +502,13 @@ private void handleSetStorePartitionCount(SetStorePartitionCount message) {
String clusterName = message.clusterName.toString();
String storeName = message.storeName.toString();
int partitionNum = message.partitionNum;

if (!admin.hasStore(clusterName, storeName)) {
LOGGER
.warn("Ignoring set-partition-count command for non-existent store {} in cluster {}", storeName, clusterName);
return;
}

admin.setStorePartitionCount(clusterName, storeName, partitionNum);

LOGGER.info("Set store: {} partition number to {} in cluster: {}", storeName, partitionNum, clusterName);
Expand All @@ -451,6 +518,11 @@ private void handleSetStore(UpdateStore message) {
String clusterName = message.clusterName.toString();
String storeName = message.storeName.toString();

if (!admin.hasStore(clusterName, storeName)) {
LOGGER.warn("Ignoring update-store command for non-existent store {} in cluster {}", storeName, clusterName);
return;
}

UpdateStoreQueryParams params = new UpdateStoreQueryParams().setOwner(message.owner.toString())
.setEnableReads(message.enableReads)
.setEnableWrites(message.enableWrites)
Expand Down

0 comments on commit ba156c7

Please sign in to comment.