minal-kyada commented on code in PR #4556: URL: https://github.com/apache/cassandra/pull/4556#discussion_r2795889413
########## src/java/org/apache/cassandra/db/WriteThresholds.java: ########## @@ -0,0 +1,121 @@ +/* + * 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.cassandra.db; + +import java.util.concurrent.TimeUnit; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.config.DataStorageSpec; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.partitions.PartitionUpdate; +import org.apache.cassandra.net.ParamType; +import org.apache.cassandra.schema.Schema; +import org.apache.cassandra.schema.TableId; +import org.apache.cassandra.schema.TableMetadata; +import org.apache.cassandra.utils.NoSpamLogger; + +/** + * Utility class for checking write threshold warnings on replicas. + * CASSANDRA-17258: paxos and accord do complex thread hand off and custom write logic which makes this patch complex, so was deferred + */ +public class WriteThresholds +{ + private static final Logger logger = LoggerFactory.getLogger(WriteThresholds.class); + private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 1, TimeUnit.MINUTES); + + /** + * Check write thresholds for all partition updates in a mutation. + * This method iterates through all partition updates in the mutation. + * + * @param mutation the mutation containing one or more partition updates + */ + public static void checkWriteThresholds(Mutation mutation) + { + if (!DatabaseDescriptor.isDaemonInitialized() || !DatabaseDescriptor.getWriteThresholdsEnabled()) + return; + + DataStorageSpec.LongBytesBound sizeWarnThreshold = DatabaseDescriptor.getWriteSizeWarnThreshold(); + int tombstoneWarnThreshold = DatabaseDescriptor.getWriteTombstoneWarnThreshold(); + + if (sizeWarnThreshold == null && tombstoneWarnThreshold == -1) + return; + + long sizeWarnBytes = sizeWarnThreshold != null ? sizeWarnThreshold.toBytes() : -1; + + for (PartitionUpdate update : mutation.getPartitionUpdates()) Review Comment: ` public static void checkWriteThresholds(Mutation mutation) { if (!DatabaseDescriptor.isDaemonInitialized() || !DatabaseDescriptor.getWriteThresholdsEnabled()) return; DataStorageSpec.LongBytesBound sizeWarnThreshold = DatabaseDescriptor.getWriteSizeWarnThreshold(); int tombstoneWarnThreshold = DatabaseDescriptor.getWriteTombstoneWarnThreshold(); if (sizeWarnThreshold == null && tombstoneWarnThreshold == -1) return; long sizeWarnBytes = sizeWarnThreshold != null ? sizeWarnThreshold.toBytes() : -1; DecoratedKey key = mutation.key(); boolean warnedSize = false; boolean warnedTombstone = false; for (TableId tableId : mutation.getTableIds()) { ColumnFamilyStore cfs = Schema.instance.getColumnFamilyStoreInstance(tableId); if (cfs == null || cfs.topPartitions == null) continue; TableMetadata metadata = cfs.metadata(); if (sizeWarnBytes != -1 && !warnedSize) { long estimatedSize = cfs.topPartitions.topSizes().getEstimate(key); if (estimatedSize > sizeWarnBytes) { MessageParams.add(ParamType.WRITE_SIZE_WARN, estimatedSize); noSpamLogger.warn("Write to {} partition {} triggered size warning; " + "estimated size is {} bytes, threshold is {} bytes (see write_size_warn_threshold)", metadata, metadata.partitionKeyType.toCQLString(key.getKey()), estimatedSize, sizeWarnBytes); warnedSize = true; } } if (tombstoneWarnThreshold != -1 && !warnedTombstone) { long estimatedTombstones = cfs.topPartitions.topTombstones().getEstimate(key); if (estimatedTombstones > tombstoneWarnThreshold) { MessageParams.add(ParamType.WRITE_TOMBSTONE_WARN, (int) estimatedTombstones); noSpamLogger.warn("Write to {} partition {} triggered tombstone warning; " + "estimated tombstone count is {}, threshold is {} (see write_tombstone_warn_threshold)", metadata, metadata.partitionKeyType.toCQLString(key.getKey()), estimatedTombstones, tombstoneWarnThreshold); warnedTombstone = true; } } } }` I was wondering: once it warns for any table, the flag is set and it won't warn for subsequent tables. However, in the existing code, we ensure that we always report the worst offender across all partition updates. I thought that if a mutation touches multiple tables, we'd want to know about the biggest problem. If the goal is just to warn once irrespective of who the culprit is, then yes, we can proceed with these changes. Thoughts? -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
