forked from apache/kafka
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[KAFKA-13369] Follower fetch protocol changes for tiered storage. (ap…
…ache#11390) This PR implements the follower fetch protocol as mentioned in KIP-405. Added a new version for ListOffsets protocol to receive local log start offset on the leader replica. This is used by follower replicas to find the local log star offset on the leader. Added a new version for FetchRequest protocol to receive OffsetMovedToTieredStorageException error. This is part of the enhanced fetch protocol as described in KIP-405. We introduced a new field locaLogStartOffset to maintain the log start offset in the local logs. Existing logStartOffset will continue to be the log start offset of the effective log that includes the segments in remote storage. When a follower receives OffsetMovedToTieredStorage, then it tries to build the required state from the leader and remote storage so that it can be ready to move to fetch state. Introduced RemoteLogManager which is responsible for initializing RemoteStorageManager and RemoteLogMetadataManager instances. receives any leader and follower replica events and partition stop events and act on them also provides APIs to fetch indexes, metadata about remote log segments. Followup PRs will add more functionality like copying segments to tiered storage, retention checks to clean local and remote log segments. This will change the local log start offset and make sure the follower fetch protocol works fine for several cases. You can look at the detailed protocol changes in KIP: https://cwiki.apache.org/confluence/display/KAFKA/KIP-405%3A+Kafka+Tiered+Storage#KIP405:KafkaTieredStorage-FollowerReplication Co-authors: [email protected], [email protected], [email protected] Reviewers: Kowshik Prakasam <[email protected]>, Cong Ding <[email protected]>, Tirtha Chatterjee <[email protected]>, Yaodong Yang <[email protected]>, Divij Vaidya <[email protected]>, Luke Chen <[email protected]>, Jun Rao <[email protected]>
- Loading branch information
Showing
65 changed files
with
3,564 additions
and
234 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
31 changes: 31 additions & 0 deletions
31
...nts/src/main/java/org/apache/kafka/common/errors/OffsetMovedToTieredStorageException.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,31 @@ | ||
| /* | ||
| * 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.common.errors; | ||
|
|
||
| public class OffsetMovedToTieredStorageException extends ApiException { | ||
|
|
||
| private static final long serialVersionUID = 1L; | ||
|
|
||
| public OffsetMovedToTieredStorageException(String message) { | ||
| super(message); | ||
| } | ||
|
|
||
| public OffsetMovedToTieredStorageException(String message, Throwable cause) { | ||
| super(message, cause); | ||
| } | ||
|
|
||
| } |
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
79 changes: 79 additions & 0 deletions
79
clients/src/main/java/org/apache/kafka/common/record/RemoteLogInputStream.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,79 @@ | ||
| /* | ||
| * 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.common.record; | ||
|
|
||
| import org.apache.kafka.common.errors.CorruptRecordException; | ||
| import org.apache.kafka.common.utils.Utils; | ||
|
|
||
| import java.io.IOException; | ||
| import java.io.InputStream; | ||
| import java.nio.ByteBuffer; | ||
|
|
||
| import static org.apache.kafka.common.record.Records.HEADER_SIZE_UP_TO_MAGIC; | ||
| import static org.apache.kafka.common.record.Records.LOG_OVERHEAD; | ||
| import static org.apache.kafka.common.record.Records.MAGIC_OFFSET; | ||
| import static org.apache.kafka.common.record.Records.SIZE_OFFSET; | ||
|
|
||
| public class RemoteLogInputStream implements LogInputStream<RecordBatch> { | ||
| private final InputStream inputStream; | ||
| // LogHeader buffer up to magic. | ||
| private final ByteBuffer logHeaderBuffer = ByteBuffer.allocate(HEADER_SIZE_UP_TO_MAGIC); | ||
|
|
||
| public RemoteLogInputStream(InputStream inputStream) { | ||
| this.inputStream = inputStream; | ||
| } | ||
|
|
||
| @Override | ||
| public RecordBatch nextBatch() throws IOException { | ||
| logHeaderBuffer.clear(); | ||
| Utils.readFully(inputStream, logHeaderBuffer); | ||
|
|
||
| if (logHeaderBuffer.position() < HEADER_SIZE_UP_TO_MAGIC) | ||
| return null; | ||
|
|
||
| logHeaderBuffer.rewind(); | ||
| int size = logHeaderBuffer.getInt(SIZE_OFFSET); | ||
|
|
||
| // V0 has the smallest overhead, stricter checking is done later | ||
| if (size < LegacyRecord.RECORD_OVERHEAD_V0) | ||
| throw new CorruptRecordException(String.format("Found record size %d smaller than minimum record " + | ||
| "overhead (%d).", size, LegacyRecord.RECORD_OVERHEAD_V0)); | ||
|
|
||
| // Total size is: "LOG_OVERHEAD + the size of the rest of the content" | ||
| int bufferSize = LOG_OVERHEAD + size; | ||
| // buffer contains the complete payload including header and records. | ||
| ByteBuffer buffer = ByteBuffer.allocate(bufferSize); | ||
|
|
||
| // write log header into buffer | ||
| buffer.put(logHeaderBuffer); | ||
|
|
||
| // write the records payload into the buffer | ||
| Utils.readFully(inputStream, buffer); | ||
| if (buffer.position() != bufferSize) | ||
| return null; | ||
| buffer.rewind(); | ||
|
|
||
| byte magic = logHeaderBuffer.get(MAGIC_OFFSET); | ||
| MutableRecordBatch batch; | ||
| if (magic > RecordBatch.MAGIC_VALUE_V1) | ||
| batch = new DefaultRecordBatch(buffer); | ||
| else | ||
| batch = new AbstractLegacyRecordBatch.ByteBufferLegacyRecordBatch(buffer); | ||
|
|
||
| return batch; | ||
| } | ||
| } |
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
clients/src/main/java/org/apache/kafka/common/utils/ChildFirstClassLoader.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 @@ | ||
| /* | ||
| * 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.common.utils; | ||
|
|
||
| import org.apache.kafka.common.KafkaException; | ||
|
|
||
| import java.io.File; | ||
| import java.io.IOException; | ||
| import java.net.URL; | ||
| import java.net.URLClassLoader; | ||
| import java.util.ArrayList; | ||
| import java.util.Enumeration; | ||
| import java.util.Locale; | ||
| import java.util.NoSuchElementException; | ||
|
|
||
| /** | ||
| * A class loader that looks for classes and resources in a specified class path first, before delegating to its parent | ||
| * class loader. | ||
| */ | ||
| public class ChildFirstClassLoader extends URLClassLoader { | ||
| static { | ||
| ClassLoader.registerAsParallelCapable(); | ||
| } | ||
|
|
||
| /** | ||
| * @param classPath Class path string | ||
| * @param parent The parent classloader. If the required class / resource cannot be found in the given classPath, | ||
| * this classloader will be used to find the class / resource. | ||
| */ | ||
| public ChildFirstClassLoader(String classPath, ClassLoader parent) { | ||
| super(classpathToURLs(classPath), parent); | ||
| } | ||
|
|
||
| static private URL[] classpathToURLs(String classPath) { | ||
| ArrayList<URL> urls = new ArrayList<>(); | ||
| for (String path : classPath.split(File.pathSeparator)) { | ||
| if (path == null || path.trim().isEmpty()) | ||
| continue; | ||
| File file = new File(path); | ||
|
|
||
| try { | ||
| if (path.endsWith("/*")) { | ||
| File parent = new File(new File(file.getCanonicalPath()).getParent()); | ||
| if (parent.isDirectory()) { | ||
| File[] files = parent.listFiles((dir, name) -> { | ||
| String lower = name.toLowerCase(Locale.ROOT); | ||
| return lower.endsWith(".jar") || lower.endsWith(".zip"); | ||
| }); | ||
| if (files != null) { | ||
| for (File jarFile : files) { | ||
| urls.add(jarFile.getCanonicalFile().toURI().toURL()); | ||
| } | ||
| } | ||
| } | ||
| } else if (file.exists()) { | ||
| urls.add(file.getCanonicalFile().toURI().toURL()); | ||
| } | ||
| } catch (IOException e) { | ||
| throw new KafkaException(e); | ||
| } | ||
| } | ||
| return urls.toArray(new URL[0]); | ||
| } | ||
|
|
||
| @Override | ||
| protected Class<?> loadClass(String name, boolean resolve) throws ClassNotFoundException { | ||
| synchronized (getClassLoadingLock(name)) { | ||
| Class<?> c = findLoadedClass(name); | ||
|
|
||
| if (c == null) { | ||
| try { | ||
| c = findClass(name); | ||
| } catch (ClassNotFoundException e) { | ||
| // Try parent | ||
| c = super.loadClass(name, false); | ||
| } | ||
| } | ||
|
|
||
| if (resolve) | ||
| resolveClass(c); | ||
|
|
||
| return c; | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public URL getResource(String name) { | ||
| URL url = findResource(name); | ||
| if (url == null) { | ||
| // try parent | ||
| url = super.getResource(name); | ||
| } | ||
| return url; | ||
| } | ||
|
|
||
| @Override | ||
| public Enumeration<URL> getResources(String name) throws IOException { | ||
| Enumeration<URL> urls1 = findResources(name); | ||
| Enumeration<URL> urls2 = getParent() != null ? getParent().getResources(name) : null; | ||
|
|
||
| return new Enumeration<URL>() { | ||
| @Override | ||
| public boolean hasMoreElements() { | ||
| return (urls1 != null && urls1.hasMoreElements()) || (urls2 != null && urls2.hasMoreElements()); | ||
| } | ||
|
|
||
| @Override | ||
| public URL nextElement() { | ||
| if (urls1 != null && urls1.hasMoreElements()) | ||
| return urls1.nextElement(); | ||
| if (urls2 != null && urls2.hasMoreElements()) | ||
| return urls2.nextElement(); | ||
| throw new NoSuchElementException(); | ||
| } | ||
| }; | ||
| } | ||
| } |
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
Oops, something went wrong.