diff --git a/src/java/org/apache/nutch/crawl/URLPartitioner.java b/src/java/org/apache/nutch/crawl/URLPartitioner.java
index ebc1534..b11257b 100644
--- a/src/java/org/apache/nutch/crawl/URLPartitioner.java
+++ b/src/java/org/apache/nutch/crawl/URLPartitioner.java
@@ -21,6 +21,12 @@ import java.net.InetAddress;
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.net.UnknownHostException;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.Map.Entry;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -55,6 +61,13 @@ public class URLPartitioner implements Configurable {
   private int seed;
   private URLNormalizers normalizers;
   private String mode = PARTITION_MODE_HOST;
+  
+  //Reduce task number is key and its count is value
+  public static TreeMap<Integer, Integer> reduceListWithCount = new TreeMap<Integer, Integer>();
+  //Url's name is key and reduce task number is value
+  public static Map<String, Integer> urlMap = new HashMap<String, Integer>();
+  public int reduceTaskNum;
+  public static final String DB_FETCH_ADAPTIVE_QUEUE_SIZE = "db.fetch.adaptive.queue.size";
 
   @Override
   public Configuration getConf() {
@@ -112,6 +125,84 @@ public class URLPartitioner implements Configurable {
     return (hashCode & Integer.MAX_VALUE) % numReduceTasks;
   }
   
+    // if "db.fetch.adaptive.queue.size" is active this method is in use.
+	public int adaptiveQueuePartitioner(String urlString, int reduceTasksCount) {
+		if(reduceTasksCount ==1){
+			return 0;
+		}
+		
+		String resultValue = null;
+		URL url = null;
+	    try {
+	      urlString = normalizers.normalize(urlString, URLNormalizers.SCOPE_PARTITION);
+	      url = new URL(urlString);
+	    } catch (MalformedURLException e) {
+	      LOG.warn("Malformed URL: '" + urlString + "'");
+	    }
+	    
+	    if (url != null) {
+	        if (mode.equals(PARTITION_MODE_HOST)) {
+	          resultValue = url.getHost();
+	        } else if (mode.equals(PARTITION_MODE_DOMAIN)) {
+	          resultValue = URLUtil.getDomainName(url);
+	        } else { // MODE IP
+	          try {
+	            InetAddress address = InetAddress.getByName(url.getHost());
+	            resultValue = address.getHostAddress();
+	          } catch (UnknownHostException e) {
+	            GeneratorJob.LOG.info("Couldn't find IP for host: " + url.getHost());
+	          }
+	        }
+	      }
+
+		//initially both of maps are empty. Url is assigned to 0. reduce task and reduce task count is set to 1
+		if (reduceListWithCount.isEmpty() && urlMap.isEmpty()) {
+			reduceTaskNum = 0;
+			urlMap.put(resultValue, reduceTaskNum);
+			reduceListWithCount.put(reduceTaskNum, 1);
+			return reduceTaskNum;
+		}
+
+		//if url exist, return its value(reduce task number). if this task number does not exist in reduceListWithCount, add to it ant set count to 1
+		if (urlMap.get(resultValue) != null) {
+			reduceTaskNum = urlMap.get(resultValue);
+			if (reduceListWithCount.get(reduceTaskNum) == null)
+				reduceListWithCount.put(reduceTaskNum, 1);
+			return reduceTaskNum;
+		}
+
+		//if url does not exist in urlMap and reduce count is more than reduce tasks count which pair of urls in urlMap, add url to urlMap with next reduce task number. 
+		//And set this reduce task number to 1.
+		if (reduceListWithCount.size() < reduceTasksCount) {
+			reduceTaskNum = reduceListWithCount.lastKey() + 1;
+			urlMap.put(resultValue, reduceListWithCount.lastKey() + 1);
+			reduceListWithCount.put(reduceTaskNum, 1);
+			return reduceTaskNum;
+		}
+
+		//if url does not exist, put it to reduce task which count is minimum.
+		reduceTaskNum = getReduceTaskNum(reduceListWithCount);
+		urlMap.put(resultValue, reduceTaskNum);
+		reduceListWithCount.put(reduceTaskNum,
+				reduceListWithCount.get(reduceTaskNum) + 1);
+
+		return reduceTaskNum;
+	}
+	
+	//Return reduce task number which has minimum count.
+	private int getReduceTaskNum(TreeMap<Integer, Integer> map) {
+
+		return Collections.min(map.entrySet(),
+				new Comparator<Map.Entry<Integer, Integer>>() {
+					@Override
+					public int compare(Entry<Integer, Integer> o1,
+							Entry<Integer, Integer> o2) {
+						return o1.getValue().intValue()
+								- o2.getValue().intValue();
+					}
+				}).getKey();
+	}
+  
   
   public static class SelectorEntryPartitioner 
       extends Partitioner<SelectorEntry, WebPage> implements Configurable {
@@ -139,11 +230,14 @@ public class URLPartitioner implements Configurable {
       extends Partitioner<IntWritable, FetchEntry> implements Configurable {
     private URLPartitioner partitioner = new URLPartitioner();
     private Configuration conf;
+    private boolean adeptiveFetchSize;
     
     @Override
     public int getPartition(IntWritable intWritable, FetchEntry fetchEntry, int numReduces) {
       String key = fetchEntry.getKey();
       String url = TableUtil.unreverseUrl(key);
+      if(adeptiveFetchSize)
+    	  return partitioner.adaptiveQueuePartitioner(url, numReduces);
       return partitioner.getPartition(url, numReduces);
     }
     
@@ -156,6 +250,7 @@ public class URLPartitioner implements Configurable {
     public void setConf(Configuration conf) {
       this.conf=conf;
       partitioner.setConf(conf);
+      adeptiveFetchSize = conf.getBoolean(DB_FETCH_ADAPTIVE_QUEUE_SIZE, false);
     }
   }