// Path: src/main/java/kafka/common/TopicAndPartition.java
// public class TopicAndPartition {
//     public final String topic;
//     public final int partition;
// 
//     public TopicAndPartition(String topic, int partition) {
//         this.topic = topic;
//         this.partition = partition;
//     }
// 
//     public TopicAndPartition(Tuple2<String, Integer> tuple) {
//         this(tuple._1, tuple._2);
//     }
// 
//     public TopicAndPartition(Partition partition) {
//         this(partition.topic, partition.partitionId);
//     }
// 
//     public TopicAndPartition(Replica replica) {
//         this(replica.topic, replica.partitionId);
//     }
// 
//     public Tuple2<String, Integer> asTuple() {
//         return Tuple2.make(topic, partition);
//     }
// 
//     @Override
//     public String toString() {
//         return String.format("[%s,%d]", topic, partition);
//     }
// 
//     @Override
//     public boolean equals(Object o) {
//         if (this == o) return true;
//         if (o == null || getClass() != o.getClass()) return false;
// 
//         TopicAndPartition that = (TopicAndPartition) o;
// 
//         if (partition != that.partition) return false;
//         if (topic != null ? !topic.equals(that.topic) : that.topic != null) return false;
// 
//         return true;
//     }
// 
//     @Override
//     public int hashCode() {
//         int result = topic != null ? topic.hashCode() : 0;
//         result = 31 * result + partition;
//         return result;
//     }
// }


import com.google.common.base.Predicate;
import com.google.common.collect.HashMultiset;
import com.google.common.collect.Multiset;
import com.google.common.collect.Table;
import kafka.common.KafkaException;
import kafka.common.TopicAndPartition;
import kafka.server.OffsetCheckpoint;
import kafka.utils.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
                      Scheduler scheduler,
                      Time time) {
        this.logDirs = logDirs;
        this.topicConfigs = topicConfigs;
        this.defaultConfig = defaultConfig;
        this.cleanerConfig = cleanerConfig;
        this.flushCheckMs = flushCheckMs;
        this.flushCheckpointMs = flushCheckpointMs;
        this.retentionCheckMs = retentionCheckMs;
        this.scheduler = scheduler;
        this.time = time;

        createAndValidateLogDirs(logDirs);
        dirLocks = lockLogDirs(logDirs);

        recoveryPointCheckpoints = Utils.map(logDirs, new Function1<File, Tuple2<File, OffsetCheckpoint>>() {
            @Override
            public Tuple2<File, OffsetCheckpoint> apply(File dir) {
                return Tuple2.make(dir, new OffsetCheckpoint(new File(dir, RecoveryPointCheckpointFile)));
            }
        });

        loadLogs(logDirs);
        cleaner = cleanerConfig.enableCleaner ? new LogCleaner(cleanerConfig, logDirs, logs, time) : null;
    }

    public static final String RecoveryPointCheckpointFile = "recovery-point-offset-checkpoint";
    public static final String LockFile = ".lock";
    public int InitialTaskDelayMs = 30 * 1000;
    private Object logCreationLock = new Object();
// Path: src/main/java/kafka/common/TopicAndPartition.java
// public class TopicAndPartition {
//     public final String topic;
//     public final int partition;
// 
//     public TopicAndPartition(String topic, int partition) {
//         this.topic = topic;
//         this.partition = partition;
//     }
// 
//     public TopicAndPartition(Tuple2<String, Integer> tuple) {
//         this(tuple._1, tuple._2);
//     }
// 
//     public TopicAndPartition(Partition partition) {
//         this(partition.topic, partition.partitionId);
//     }
// 
//     public TopicAndPartition(Replica replica) {
//         this(replica.topic, replica.partitionId);
//     }
// 
//     public Tuple2<String, Integer> asTuple() {
//         return Tuple2.make(topic, partition);
//     }
// 
//     @Override
//     public String toString() {
//         return String.format("[%s,%d]", topic, partition);
//     }
// 
//     @Override
//     public boolean equals(Object o) {
//         if (this == o) return true;
//         if (o == null || getClass() != o.getClass()) return false;
// 
//         TopicAndPartition that = (TopicAndPartition) o;
// 
//         if (partition != that.partition) return false;
//         if (topic != null ? !topic.equals(that.topic) : that.topic != null) return false;
// 
//         return true;
//     }
// 
//     @Override
//     public int hashCode() {
//         int result = topic != null ? topic.hashCode() : 0;
//         result = 31 * result + partition;
//         return result;
//     }
// }


// Path: src/main/java/kafka/log/LogManager.java
import com.google.common.base.Predicate;
import com.google.common.collect.HashMultiset;
import com.google.common.collect.Multiset;
import com.google.common.collect.Table;
import kafka.common.KafkaException;
import kafka.common.TopicAndPartition;
import kafka.server.OffsetCheckpoint;
import kafka.utils.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

                      Scheduler scheduler,
                      Time time) {
        this.logDirs = logDirs;
        this.topicConfigs = topicConfigs;
        this.defaultConfig = defaultConfig;
        this.cleanerConfig = cleanerConfig;
        this.flushCheckMs = flushCheckMs;
        this.flushCheckpointMs = flushCheckpointMs;
        this.retentionCheckMs = retentionCheckMs;
        this.scheduler = scheduler;
        this.time = time;

        createAndValidateLogDirs(logDirs);
        dirLocks = lockLogDirs(logDirs);

        recoveryPointCheckpoints = Utils.map(logDirs, new Function1<File, Tuple2<File, OffsetCheckpoint>>() {
            @Override
            public Tuple2<File, OffsetCheckpoint> apply(File dir) {
                return Tuple2.make(dir, new OffsetCheckpoint(new File(dir, RecoveryPointCheckpointFile)));
            }
        });

        loadLogs(logDirs);
        cleaner = cleanerConfig.enableCleaner ? new LogCleaner(cleanerConfig, logDirs, logs, time) : null;
    }

    public static final String RecoveryPointCheckpointFile = "recovery-point-offset-checkpoint";
    public static final String LockFile = ".lock";
    public int InitialTaskDelayMs = 30 * 1000;
    private Object logCreationLock = new Object();
    private Pool<TopicAndPartition, Log> logs = new Pool<TopicAndPartition, Log>();