diff --git a/metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java index 319c9d537eec1..b197518391622 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java +++ b/metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java @@ -324,8 +324,9 @@ public void replay(FeatureLevelRecord record) { "supports versions " + range); } if (record.name().equals(MetadataVersion.FEATURE_NAME)) { - log.info("Setting metadata.version to {}", record.featureLevel()); - metadataVersion.set(MetadataVersion.fromFeatureLevel(record.featureLevel())); + MetadataVersion mv = MetadataVersion.fromFeatureLevel(record.featureLevel()); + log.info("Setting metadata version to {}", mv); + metadataVersion.set(mv); } else { if (record.featureLevel() == 0) { log.info("Removing feature {}", record.name()); diff --git a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java index 163939835430b..66fc2611a3d58 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java +++ b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java @@ -88,6 +88,7 @@ import org.apache.kafka.metadata.migration.ZkRecordConsumer; import org.apache.kafka.metadata.placement.ReplicaPlacer; import org.apache.kafka.metadata.placement.StripedReplicaPlacer; +import org.apache.kafka.metadata.util.RecordRedactor; import org.apache.kafka.deferred.DeferredEventQueue; import org.apache.kafka.deferred.DeferredEvent; import org.apache.kafka.queue.EventQueue.EarliestDeadlineFunction; @@ -106,6 +107,7 @@ import org.apache.kafka.server.policy.AlterConfigPolicy; import org.apache.kafka.server.policy.CreateTopicPolicy; import org.apache.kafka.snapshot.SnapshotReader; +import org.apache.kafka.snapshot.Snapshots; import org.apache.kafka.timeline.SnapshotRegistry; import org.slf4j.Logger; @@ -130,7 +132,6 @@ import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; -import java.util.stream.Collectors; import static java.util.concurrent.TimeUnit.MICROSECONDS; import static java.util.concurrent.TimeUnit.NANOSECONDS; @@ -963,15 +964,8 @@ public void handleCommit(BatchReader reader) { // If the controller is a standby, replay the records that were // created by the active controller. if (log.isDebugEnabled()) { - if (log.isTraceEnabled()) { - log.trace("Replaying commits from the active node up to " + - "offset {} and epoch {}: {}.", offset, epoch, messages.stream() - .map(ApiMessageAndVersion::toString) - .collect(Collectors.joining(", "))); - } else { - log.debug("Replaying commits from the active node up to " + - "offset {} and epoch {}.", offset, epoch); - } + log.debug("Replaying commits from the active node up to " + + "offset {} and epoch {}.", offset, epoch); } int i = 1; for (ApiMessageAndVersion message : messages) { @@ -1011,13 +1005,13 @@ public void handleCommit(BatchReader reader) { public void handleSnapshot(SnapshotReader reader) { appendRaftEvent(String.format("handleSnapshot[snapshotId=%s]", reader.snapshotId()), () -> { try { + String snapshotName = Snapshots.filenameFromSnapshotId(reader.snapshotId()); if (isActiveController()) { - throw fatalFaultHandler.handleFault(String.format("Asked to load snapshot " + - "(%s) when it is the active controller (%d)", reader.snapshotId(), - curClaimEpoch)); + throw fatalFaultHandler.handleFault("Asked to load snapshot " + snapshotName + + ", but we are the active controller at epoch " + curClaimEpoch); } - log.info("Starting to replay snapshot ({}), from last commit offset ({}) and epoch ({})", - reader.snapshotId(), lastCommittedOffset, lastCommittedEpoch); + log.info("Starting to replay snapshot {}, from last commit offset {} and epoch {}", + snapshotName, lastCommittedOffset, lastCommittedEpoch); resetToEmptyState(); @@ -1026,16 +1020,8 @@ public void handleSnapshot(SnapshotReader reader) { long offset = batch.lastOffset(); List messages = batch.records(); - if (log.isDebugEnabled()) { - if (log.isTraceEnabled()) { - log.trace("Replaying snapshot ({}) batch with last offset of {}: {}", - reader.snapshotId(), offset, messages.stream().map(ApiMessageAndVersion::toString). - collect(Collectors.joining(", "))); - } else { - log.debug("Replaying snapshot ({}) batch with last offset of {}", - reader.snapshotId(), offset); - } - } + log.debug("Replaying snapshot {} batch with last offset of {}", + snapshotName, offset); int i = 1; for (ApiMessageAndVersion message : messages) { @@ -1052,6 +1038,7 @@ public void handleSnapshot(SnapshotReader reader) { i++; } } + log.info("Finished replaying snapshot {}", snapshotName); updateLastCommittedState( reader.lastContainedLogOffset(), @@ -1509,6 +1496,16 @@ private void handleFeatureControlChange() { * if this record is from a snapshot, this is used along with RegisterBrokerRecord */ private void replay(ApiMessage message, Optional snapshotId, long batchLastOffset) { + if (log.isTraceEnabled()) { + if (snapshotId.isPresent()) { + log.trace("Replaying snapshot {} record {}", + Snapshots.filenameFromSnapshotId(snapshotId.get()), + recordRedactor.toLoggableString(message)); + } else { + log.trace("Replaying log record {} with batchLastOffset {}", + recordRedactor.toLoggableString(message), batchLastOffset); + } + } logReplayTracker.replay(message); MetadataRecordType type = MetadataRecordType.fromId(message.apiKey()); switch (type) { @@ -1780,11 +1777,6 @@ private enum ImbalanceSchedule { */ private boolean noOpRecordScheduled = false; - /** - * Tracks if a snapshot generate was scheduled. - */ - private boolean generateSnapshotScheduled = false; - /** * The bootstrap metadata to use for initialization if needed. */ @@ -1799,6 +1791,10 @@ private enum ImbalanceSchedule { */ private final int maxRecordsPerBatch; + /** + * Supports converting records to strings without disclosing passwords. + */ + private final RecordRedactor recordRedactor; private QuorumController( FaultHandler fatalFaultHandler, @@ -1835,7 +1831,7 @@ private QuorumController( this.time = time; this.controllerMetrics = controllerMetrics; this.snapshotRegistry = new SnapshotRegistry(logContext); - this.deferredEventQueue = new DeferredEventQueue(); + this.deferredEventQueue = new DeferredEventQueue(logContext); this.resourceExists = new ConfigResourceExistenceChecker(); this.configurationControl = new ConfigurationControlManager.Builder(). setLogContext(logContext). @@ -1901,11 +1897,13 @@ private QuorumController( this.needToCompleteAuthorizerLoad = authorizer.isPresent(); this.zkRecordConsumer = new MigrationRecordConsumer(); this.zkMigrationEnabled = zkMigrationEnabled; + this.recordRedactor = new RecordRedactor(configSchema); updateWriteOffset(-1); resetToEmptyState(); - log.info("Creating new QuorumController with clusterId {}, authorizer {}.", clusterId, authorizer); + log.info("Creating new QuorumController with clusterId {}, authorizer {}.{}", + clusterId, authorizer, zkMigrationEnabled ? " ZK migration mode is enabled." : ""); this.raftClient.register(metaLogListener); } diff --git a/metadata/src/main/java/org/apache/kafka/metadata/util/RecordRedactor.java b/metadata/src/main/java/org/apache/kafka/metadata/util/RecordRedactor.java new file mode 100644 index 0000000000000..d131b643c6be9 --- /dev/null +++ b/metadata/src/main/java/org/apache/kafka/metadata/util/RecordRedactor.java @@ -0,0 +1,63 @@ +/* + * 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.metadata.util; + +import org.apache.kafka.common.metadata.ConfigRecord; +import org.apache.kafka.common.metadata.MetadataRecordType; +import org.apache.kafka.common.metadata.UserScramCredentialRecord; +import org.apache.kafka.common.protocol.ApiMessage; +import org.apache.kafka.metadata.KafkaConfigSchema; + + +/** + * Converts a metadata record to a string suitable for logging to slf4j. + * This means that passwords and key material are omitted from the output. + */ +public final class RecordRedactor { + private final KafkaConfigSchema configSchema; + + public RecordRedactor(KafkaConfigSchema configSchema) { + this.configSchema = configSchema; + } + + public String toLoggableString(ApiMessage message) { + MetadataRecordType type = MetadataRecordType.fromId(message.apiKey()); + switch (type) { + case CONFIG_RECORD: { + if (!configSchema.isSensitive((ConfigRecord) message)) { + return message.toString(); + } + ConfigRecord duplicate = ((ConfigRecord) message).duplicate(); + duplicate.setValue("(redacted)"); + return duplicate.toString(); + } + case USER_SCRAM_CREDENTIAL_RECORD: { + UserScramCredentialRecord record = (UserScramCredentialRecord) message; + return "UserScramCredentialRecord(" + + "name=" + record.name() + + ", mechanism=" + record.mechanism() + + ", salt=(redacted)" + + ", saltedPassword=(redacted)" + + ", iterations=" + record.iterations() + + ")"; + } + default: + return message.toString(); + } + } +} diff --git a/metadata/src/main/java/org/apache/kafka/metadata/util/SnapshotReason.java b/metadata/src/main/java/org/apache/kafka/metadata/util/SnapshotReason.java deleted file mode 100644 index a65bdd5593b63..0000000000000 --- a/metadata/src/main/java/org/apache/kafka/metadata/util/SnapshotReason.java +++ /dev/null @@ -1,60 +0,0 @@ -/* - * 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.metadata.util; - -import java.util.Collection; -import org.apache.kafka.common.utils.Utils; -import org.apache.kafka.server.common.MetadataVersion; - -/** - * Reason for generating a snapshot. - */ -public final class SnapshotReason { - static public final SnapshotReason UNKNOWN = new SnapshotReason("unknown reason"); - - static public SnapshotReason maxBytesExceeded(long bytes, long maxBytes) { - return new SnapshotReason(String.format("%s bytes exceeded the maximum bytes of %s", bytes, maxBytes)); - } - - static public SnapshotReason maxIntervalExceeded(long interval, long maxInterval) { - return new SnapshotReason( - String.format("%s ms exceeded the maximum snapshot interval of %s ms", interval, maxInterval) - ); - } - - static public SnapshotReason metadataVersionChanged(MetadataVersion metadataVersion) { - return new SnapshotReason(String.format("metadata version was changed to %s", metadataVersion)); - } - - /** - * Converts a collection of reasons into a string. - */ - static public String stringFromReasons(Collection reasons) { - return Utils.join(reasons, ", "); - } - - private final String reason; - - private SnapshotReason(String reason) { - this.reason = reason; - } - - @Override - public String toString() { - return reason; - } -} diff --git a/metadata/src/test/java/org/apache/kafka/metadata/util/RecordRedactorTest.java b/metadata/src/test/java/org/apache/kafka/metadata/util/RecordRedactorTest.java new file mode 100644 index 0000000000000..a4e3bf8f5415d --- /dev/null +++ b/metadata/src/test/java/org/apache/kafka/metadata/util/RecordRedactorTest.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.metadata.util; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.config.ConfigDef; +import org.apache.kafka.common.config.ConfigResource; +import org.apache.kafka.common.metadata.ConfigRecord; +import org.apache.kafka.common.metadata.TopicRecord; +import org.apache.kafka.common.metadata.UserScramCredentialRecord; +import org.apache.kafka.metadata.KafkaConfigSchema; +import org.junit.jupiter.api.Test; + +import static org.apache.kafka.common.config.ConfigResource.Type.BROKER; +import static org.junit.jupiter.api.Assertions.assertEquals; + + +final public class RecordRedactorTest { + public static final Map CONFIGS = new HashMap<>(); + + static { + CONFIGS.put(BROKER, new ConfigDef(). + define("foobar", ConfigDef.Type.LIST, "1", ConfigDef.Importance.HIGH, "foo bar doc"). + define("quux", ConfigDef.Type.PASSWORD, ConfigDef.Importance.HIGH, "quuux2 doc")); + } + + private static final KafkaConfigSchema SCHEMA = new KafkaConfigSchema(CONFIGS, Collections.emptyMap()); + + private static final RecordRedactor REDACTOR = new RecordRedactor(SCHEMA); + + @Test + public void testTopicRecordToString() { + assertEquals("TopicRecord(name='foo', topicId=UOovKkohSU6AGdYW33ZUNg)", + REDACTOR.toLoggableString(new TopicRecord(). + setTopicId(Uuid.fromString("UOovKkohSU6AGdYW33ZUNg")). + setName("foo"))); + } + + @Test + public void testUserScramCredentialRecordToString() { + assertEquals("UserScramCredentialRecord(name=bob, mechanism=0, " + + "salt=(redacted), serverKey=(redacted), storedKey=(redacted), iterations=128)", + REDACTOR.toLoggableString(new UserScramCredentialRecord(). + setName("bob"). + setMechanism((byte) 0). + setSalt(new byte[512]). + setServerKey(new byte[128]). + setStoredKey(new byte[128]). + setIterations(128))); + } + + @Test + public void testSensitiveConfigRecordToString() { + assertEquals("ConfigRecord(resourceType=4, resourceName='0', name='quux', " + + "value='(redacted)')", + REDACTOR.toLoggableString(new ConfigRecord(). + setResourceType(BROKER.id()). + setResourceName("0"). + setName("quux"). + setValue("mysecret"))); + } + + @Test + public void testNonSensitiveConfigRecordToString() { + assertEquals("ConfigRecord(resourceType=4, resourceName='0', name='foobar', " + + "value='item1,item2')", + REDACTOR.toLoggableString(new ConfigRecord(). + setResourceType(BROKER.id()). + setResourceName("0"). + setName("foobar"). + setValue("item1,item2"))); + } +} diff --git a/metadata/src/test/java/org/apache/kafka/metadata/util/SnapshotReasonTest.java b/metadata/src/test/java/org/apache/kafka/metadata/util/SnapshotReasonTest.java deleted file mode 100644 index e90a4d2928d84..0000000000000 --- a/metadata/src/test/java/org/apache/kafka/metadata/util/SnapshotReasonTest.java +++ /dev/null @@ -1,80 +0,0 @@ -/* - * 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.metadata.util; - -import java.util.Arrays; -import java.util.List; -import org.apache.kafka.server.common.MetadataVersion; -import org.junit.jupiter.api.Test; -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertTrue; - -public final class SnapshotReasonTest { - @Test - public void testUnknown() { - assertEquals("unknown reason", SnapshotReason.UNKNOWN.toString()); - } - - @Test - public void testMaxBytesExceeded() { - long bytes = 1000; - long maxBytes = 900; - String expectedMessage = "1000 bytes exceeded the maximum bytes of 900"; - - assertEquals(expectedMessage, SnapshotReason.maxBytesExceeded(bytes, maxBytes).toString()); - } - - @Test - public void testMaxIntervalExceeded() { - long interval = 1000; - long maxInterval = 900; - String expectedMessage = "1000 ms exceeded the maximum snapshot interval of 900 ms"; - - assertEquals(expectedMessage, SnapshotReason.maxIntervalExceeded(interval, maxInterval).toString()); - } - - @Test - public void testMetadataVersionChanged() { - MetadataVersion metadataVersion = MetadataVersion.IBP_3_3_IV3; - String expectedMessage = "metadata version was changed to 3.3-IV3"; - - assertEquals(expectedMessage, SnapshotReason.metadataVersionChanged(metadataVersion).toString()); - } - - @Test - public void testJoinReasons() { - long bytes = 1000; - long maxBytes = 900; - long interval = 1000; - long maxInterval = 900; - MetadataVersion metadataVersion = MetadataVersion.IBP_3_3_IV3; - - List reasons = Arrays.asList( - SnapshotReason.UNKNOWN, - SnapshotReason.maxBytesExceeded(bytes, maxBytes), - SnapshotReason.maxIntervalExceeded(interval, maxInterval), - SnapshotReason.metadataVersionChanged(metadataVersion) - ); - - String joinedReasons = SnapshotReason.stringFromReasons(reasons); - - assertTrue(joinedReasons.contains("unknown reason"), joinedReasons); - assertTrue(joinedReasons.contains("1000 bytes exceeded the maximum bytes of 900"), joinedReasons); - assertTrue(joinedReasons.contains("1000 ms exceeded the maximum snapshot interval of 900 ms"), joinedReasons); - assertTrue(joinedReasons.contains("metadata version was changed to 3.3-IV3"), joinedReasons); - } -} diff --git a/server-common/src/main/java/org/apache/kafka/deferred/DeferredEventQueue.java b/server-common/src/main/java/org/apache/kafka/deferred/DeferredEventQueue.java index b66d0a3db8ed1..32e183ce9b025 100644 --- a/server-common/src/main/java/org/apache/kafka/deferred/DeferredEventQueue.java +++ b/server-common/src/main/java/org/apache/kafka/deferred/DeferredEventQueue.java @@ -17,6 +17,9 @@ package org.apache.kafka.deferred; +import org.apache.kafka.common.utils.LogContext; +import org.slf4j.Logger; + import java.util.ArrayList; import java.util.Iterator; import java.util.List; @@ -29,12 +32,18 @@ * We wait for the high watermark of the log to advance before completing them. */ public class DeferredEventQueue { + private final Logger log; + /** * A map from log offsets to events. Each event will be completed once the log * advances past its offset. */ private final TreeMap> pending = new TreeMap<>(); + public DeferredEventQueue(LogContext logContext) { + this.log = logContext.logger(DeferredEventQueue.class); + } + /** * Complete some purgatory entries. * @@ -42,16 +51,23 @@ public class DeferredEventQueue { */ public void completeUpTo(long offset) { Iterator>> iter = pending.entrySet().iterator(); + int numCompleted = 0; while (iter.hasNext()) { Entry> entry = iter.next(); if (entry.getKey() > offset) { break; } for (DeferredEvent event : entry.getValue()) { + log.debug("completeUpTo({}): successfully completing {}", offset, event); event.complete(null); + numCompleted++; } iter.remove(); } + if (log.isTraceEnabled()) { + log.trace("completeUpTo({}): successfully completed {} deferred entries", + offset, numCompleted); + } } /** @@ -64,6 +80,7 @@ public void failAll(Exception exception) { while (iter.hasNext()) { Entry> entry = iter.next(); for (DeferredEvent event : entry.getValue()) { + log.info("failAll({}): failing {}.", exception.getClass().getSimpleName(), event); event.complete(exception); } iter.remove(); @@ -91,6 +108,9 @@ public void add(long offset, DeferredEvent event) { pending.put(offset, events); } events.add(event); + if (log.isTraceEnabled()) { + log.trace("Adding deferred event {} at offset {}", event, offset); + } } /** diff --git a/server-common/src/test/java/org/apache/kafka/deferred/DeferredEventQueueTest.java b/server-common/src/test/java/org/apache/kafka/deferred/DeferredEventQueueTest.java index 09ec099469431..7c4f0e62a95fa 100644 --- a/server-common/src/test/java/org/apache/kafka/deferred/DeferredEventQueueTest.java +++ b/server-common/src/test/java/org/apache/kafka/deferred/DeferredEventQueueTest.java @@ -21,6 +21,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; +import org.apache.kafka.common.utils.LogContext; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; @@ -51,7 +52,7 @@ CompletableFuture future() { @Test public void testCompleteEvents() { - DeferredEventQueue deferredEventQueue = new DeferredEventQueue(); + DeferredEventQueue deferredEventQueue = new DeferredEventQueue(new LogContext()); SampleDeferredEvent event1 = new SampleDeferredEvent(); SampleDeferredEvent event2 = new SampleDeferredEvent(); SampleDeferredEvent event3 = new SampleDeferredEvent(); @@ -72,7 +73,7 @@ public void testCompleteEvents() { @Test public void testFailOnIncorrectOrdering() { - DeferredEventQueue deferredEventQueue = new DeferredEventQueue(); + DeferredEventQueue deferredEventQueue = new DeferredEventQueue(new LogContext()); SampleDeferredEvent event1 = new SampleDeferredEvent(); SampleDeferredEvent event2 = new SampleDeferredEvent(); deferredEventQueue.add(2, event1); @@ -81,7 +82,7 @@ public void testFailOnIncorrectOrdering() { @Test public void testFailEvents() { - DeferredEventQueue deferredEventQueue = new DeferredEventQueue(); + DeferredEventQueue deferredEventQueue = new DeferredEventQueue(new LogContext()); SampleDeferredEvent event1 = new SampleDeferredEvent(); SampleDeferredEvent event2 = new SampleDeferredEvent(); SampleDeferredEvent event3 = new SampleDeferredEvent();