mumrah commented on a change in pull request #10949: URL: https://github.com/apache/kafka/pull/10949#discussion_r661800484
########## File path: metadata/src/main/java/org/apache/kafka/image/ClusterDelta.java ########## @@ -0,0 +1,126 @@ +/* + * 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.kafka.image; + +import org.apache.kafka.common.metadata.BrokerRegistrationChangeRecord; +import org.apache.kafka.common.metadata.FenceBrokerRecord; +import org.apache.kafka.common.metadata.RegisterBrokerRecord; +import org.apache.kafka.common.metadata.UnfenceBrokerRecord; +import org.apache.kafka.common.metadata.UnregisterBrokerRecord; +import org.apache.kafka.metadata.BrokerRegistration; + +import java.util.HashMap; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Optional; + + +/** + * Represents changes to the cluster in the metadata image. + */ +public final class ClusterDelta { + private final ClusterImage image; + private final HashMap<Integer, Optional<BrokerRegistration>> changedBrokers = new HashMap<>(); + + public ClusterDelta(ClusterImage image) { + this.image = image; + } + + public HashMap<Integer, Optional<BrokerRegistration>> changedBrokers() { + return changedBrokers; + } + + public BrokerRegistration broker(int nodeId) { + Optional<BrokerRegistration> result = changedBrokers.get(nodeId); + if (result != null) { + return result.orElse(null); + } + return image.broker(nodeId); + } + + public void finishSnapshot() { + for (Integer brokerId : image.brokers().keySet()) { + if (!changedBrokers.containsKey(brokerId)) { + changedBrokers.put(brokerId, Optional.empty()); + } + } + } + + public void replay(RegisterBrokerRecord record) { + BrokerRegistration broker = BrokerRegistration.fromRecord(record); + changedBrokers.put(broker.id(), Optional.of(broker)); + } + + public void replay(UnregisterBrokerRecord record) { + changedBrokers.put(record.brokerId(), Optional.empty()); + } + + public void replay(FenceBrokerRecord record) { + BrokerRegistration broker = broker(record.id()); + if (broker == null) { + throw new RuntimeException("Tried to fence broker " + record.id() + Review comment: Maybe IllegalStateException for these? Shouldn't it be impossible to get into this state (aside from missing records or controller bugs)? ########## File path: metadata/src/main/java/org/apache/kafka/image/ClientQuotasDelta.java ########## @@ -0,0 +1,87 @@ +/* + * 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.kafka.image; + +import org.apache.kafka.common.metadata.ClientQuotaRecord; +import org.apache.kafka.common.quota.ClientQuotaEntity; + +import java.util.HashMap; +import java.util.Map; +import java.util.Map.Entry; + + +public final class ClientQuotasDelta { + private final ClientQuotasImage image; + private final Map<ClientQuotaEntity, ClientQuotaDelta> changes = new HashMap<>(); + + public ClientQuotasDelta(ClientQuotasImage image) { + this.image = image; + } + + public Map<ClientQuotaEntity, ClientQuotaDelta> changes() { + return changes; + } + + public void finishSnapshot() { + for (Entry<ClientQuotaEntity, ClientQuotaImage> entry : image.entities().entrySet()) { + ClientQuotaEntity entity = entry.getKey(); + ClientQuotaImage quotaImage = entry.getValue(); + ClientQuotaDelta quotaDelta = changes.computeIfAbsent(entity, + __ -> new ClientQuotaDelta(quotaImage)); + quotaDelta.finishSnapshot(); + } + } + + public void replay(ClientQuotaRecord record) { + ClientQuotaEntity entity = ClientQuotaImage.dataToEntity(record.entity()); + ClientQuotaDelta change = changes.get(entity); + if (change == null) { Review comment: I think this could be done with `computeIfAbsent` like in "finishSnapshot" above ########## File path: metadata/src/main/java/org/apache/kafka/image/ClusterDelta.java ########## @@ -0,0 +1,126 @@ +/* + * 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.kafka.image; + +import org.apache.kafka.common.metadata.BrokerRegistrationChangeRecord; +import org.apache.kafka.common.metadata.FenceBrokerRecord; +import org.apache.kafka.common.metadata.RegisterBrokerRecord; +import org.apache.kafka.common.metadata.UnfenceBrokerRecord; +import org.apache.kafka.common.metadata.UnregisterBrokerRecord; +import org.apache.kafka.metadata.BrokerRegistration; + +import java.util.HashMap; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Optional; + + +/** + * Represents changes to the cluster in the metadata image. + */ +public final class ClusterDelta { + private final ClusterImage image; + private final HashMap<Integer, Optional<BrokerRegistration>> changedBrokers = new HashMap<>(); + + public ClusterDelta(ClusterImage image) { + this.image = image; + } + + public HashMap<Integer, Optional<BrokerRegistration>> changedBrokers() { + return changedBrokers; + } + + public BrokerRegistration broker(int nodeId) { + Optional<BrokerRegistration> result = changedBrokers.get(nodeId); + if (result != null) { + return result.orElse(null); + } + return image.broker(nodeId); + } + + public void finishSnapshot() { + for (Integer brokerId : image.brokers().keySet()) { + if (!changedBrokers.containsKey(brokerId)) { + changedBrokers.put(brokerId, Optional.empty()); + } + } + } + + public void replay(RegisterBrokerRecord record) { + BrokerRegistration broker = BrokerRegistration.fromRecord(record); + changedBrokers.put(broker.id(), Optional.of(broker)); + } + + public void replay(UnregisterBrokerRecord record) { + changedBrokers.put(record.brokerId(), Optional.empty()); + } + + public void replay(FenceBrokerRecord record) { + BrokerRegistration broker = broker(record.id()); + if (broker == null) { + throw new RuntimeException("Tried to fence broker " + record.id() + + ", but that broker was not registered."); + } + changedBrokers.put(record.id(), Optional.of(broker.cloneWithFencing(true))); + } + + public void replay(UnfenceBrokerRecord record) { + BrokerRegistration broker = broker(record.id()); + if (broker == null) { + throw new RuntimeException("Tried to unfence broker " + record.id() + + ", but that broker was not registered."); + } + changedBrokers.put(record.id(), Optional.of(broker.cloneWithFencing(false))); + } + + public void replay(BrokerRegistrationChangeRecord record) { + BrokerRegistration broker = broker(record.brokerId()); + if (broker == null) { + throw new RuntimeException("Tried to change broker " + record.brokerId() + + ", but that broker was not registered."); + } + if (record.fenced() < 0) { + changedBrokers.put(record.brokerId(), Optional.of(broker.cloneWithFencing(false))); + } else if (record.fenced() > 0) { + changedBrokers.put(record.brokerId(), Optional.of(broker.cloneWithFencing(true))); + } + } + + public ClusterImage apply() { + Map<Integer, BrokerRegistration> newBrokers = new HashMap<>(image.brokers().size()); + for (Entry<Integer, BrokerRegistration> entry : image.brokers().entrySet()) { + int nodeId = entry.getKey(); + Optional<BrokerRegistration> change = changedBrokers.get(nodeId); + if (change == null) { + newBrokers.put(nodeId, entry.getValue()); + } else if (change.isPresent()) { + newBrokers.put(nodeId, change.get()); + } Review comment: It's a little unclear as to how we are removing things from the cache here. If I understand correctly, we stick an empty Optional in the cache to represent removal. And here by _not_ handling the empty optional case, we exclude the broker from the new image. But if that's the case, why do we need the Optional in the first place? Can we directly remove the broker from "changedBrokers" when handling UnregisterBrokerRecord? This question applies to some other Delta classes that are using Optional as well ########## File path: metadata/src/main/java/org/apache/kafka/image/ConfigurationsDelta.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.kafka.image; + +import org.apache.kafka.common.config.ConfigResource; +import org.apache.kafka.common.config.ConfigResource.Type; +import org.apache.kafka.common.metadata.ConfigRecord; +import org.apache.kafka.common.metadata.RemoveTopicRecord; + +import java.util.HashMap; +import java.util.Map; +import java.util.Map.Entry; + + +/** + * Represents changes to the configurations in the metadata image. + */ +public final class ConfigurationsDelta { + private final ConfigurationsImage image; + private final Map<ConfigResource, ConfigurationDelta> changes = new HashMap<>(); + + public ConfigurationsDelta(ConfigurationsImage image) { + this.image = image; + } + + public Map<ConfigResource, ConfigurationDelta> changes() { + return changes; + } + + public void finishSnapshot() { + for (Entry<ConfigResource, ConfigurationImage> entry : image.resourceData().entrySet()) { + ConfigResource resource = entry.getKey(); + ConfigurationImage configurationImage = entry.getValue(); + ConfigurationDelta configurationDelta = changes.computeIfAbsent(resource, Review comment: (very minor) nit: inconsistent naming of these in this class ########## File path: metadata/src/main/java/org/apache/kafka/image/ClientQuotaDelta.java ########## @@ -0,0 +1,76 @@ +/* + * 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.kafka.image; + +import org.apache.kafka.common.metadata.ClientQuotaRecord; + +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; + +import static java.util.Map.Entry; + + +public final class ClientQuotaDelta { + private final ClientQuotaImage image; + private final Map<String, Optional<Double>> changes = new HashMap<>(); + + public ClientQuotaDelta(ClientQuotaImage image) { + this.image = image; + } + + public Map<String, Optional<Double>> changes() { + return changes; + } + + public void finishSnapshot() { + for (String key : image.quotas().keySet()) { + if (!changes.containsKey(key)) { + changes.put(key, Optional.empty()); + } + } + } + + public void replay(ClientQuotaRecord record) { + if (record.remove()) { + changes.put(record.key(), Optional.empty()); + } else { + changes.put(record.key(), Optional.of(record.value())); + } + } + + public ClientQuotaImage apply() { + Map<String, Double> newQuotas = new HashMap<>(image.quotas().size()); + for (Entry<String, Double> entry : image.quotas().entrySet()) { + Optional<Double> change = changes.get(entry.getKey()); + if (change == null) { + newQuotas.put(entry.getKey(), entry.getValue()); + } else if (change.isPresent()) { + newQuotas.put(entry.getKey(), change.get()); + } Review comment: Are we missing the removal case here? ########## File path: metadata/src/main/java/org/apache/kafka/image/ConfigurationsImage.java ########## @@ -0,0 +1,92 @@ +/* + * 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.kafka.image; + +import org.apache.kafka.common.config.ConfigResource; +import org.apache.kafka.server.common.ApiMessageAndVersion; + +import java.io.IOException; +import java.util.Collections; +import java.util.List; +import java.util.Map.Entry; +import java.util.Map; +import java.util.Objects; +import java.util.Properties; +import java.util.function.Consumer; +import java.util.stream.Collectors; + + +/** + * Represents the configurations in the metadata image. + * + * This class is thread-safe. + */ +public final class ConfigurationsImage { + public static final ConfigurationsImage EMPTY = + new ConfigurationsImage(Collections.emptyMap()); + + private final Map<ConfigResource, ConfigurationImage> data; + + public ConfigurationsImage(Map<ConfigResource, ConfigurationImage> data) { + this.data = data; + } + + public boolean isEmpty() { + return data.isEmpty(); + } + + Map<ConfigResource, ConfigurationImage> resourceData() { + return data; + } + + public Properties configProperties(ConfigResource configResource) { Review comment: Hm, kind of annoying that we have to return Properties here, but (as far as I know) there is no way to make an immutable Properties -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
