Fixed authentication issue with remote, auth-enabled databases.

Author: djktno

.gitignore

@@ -7,3 +7,4 @@ build
 *.ipr
 out
 dist
+jvm*.properties

src/java/cascading/mongodb/MongoDBConfiguration.java

@@ -176,5 +176,26 @@ void setPort(int port)
     {
         jobConf.setInt(PORT, port);
     }
+
+    String getUsername()
+    {
+        return jobConf.get(USERNAME, null);
+    }
+
+    void setUsername(String username)
+    {
+        jobConf.set(USERNAME, username);
+    }
+
+    char[] getPassword()
+    {
+        String s = jobConf.get(PASSWORD, null);
+        return s.toCharArray();
+    }
+
+    void setPassword(char[] password)
+    {
+        jobConf.set(PASSWORD, new String(password));
+    }
 
 }

src/java/cascading/mongodb/MongoDBOutputFormat.java

@@ -1,26 +1,20 @@
 package cascading.mongodb;
 
-import cascading.tuple.Tuple;
-import com.mongodb.BasicDBObject;
 import com.mongodb.DB;
-import com.mongodb.DBCollection;
-import com.mongodb.MongoException;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.OutputFormat;
 import org.apache.hadoop.mapred.RecordWriter;
-import org.apache.hadoop.mapred.Reporter;
 import org.apache.hadoop.util.Progressable;
 import org.apache.log4j.Logger;
-import org.bson.types.ObjectId;
 
 import java.io.IOException;
 
 /**
  * An OutputFormat that sends reduce output to a MongoDB collection.
  * <p/>
  * {@link MongoDBOutputFormat} accepts &lt;key, value&gt; pairs, where
- * the key has a type extending MongoDBWritable.  Returned {@link MongoDBRecordWriter}
+ * the key has a type extending MongoDBWritable.  Returned {@link cascading.mongodb.DefaultMongoDBRecordWriter}
  * writes a document to the collection with the Fields names specified as the
  * attributes of the document. 
  * 
@@ -36,7 +30,17 @@ public RecordWriter<K, V> getRecordWriter(FileSystem fileSystem, JobConf jobConf
         String collection = dbConfiguration.getCollection();
         DB db = dbConfiguration.getDB();
 
-        return new MongoDBRecordWriter(db, collection);
+
+        String username = dbConfiguration.getUsername();
+        if (username == null && "".equals(username))
+            log.warn("Username was not set.  If your database has security enabled, your upcoming writes will fail.  If not, no worries!");
+
+        else if (!db.authenticate(dbConfiguration.getUsername(), dbConfiguration.getPassword()))
+        {
+            throw new IllegalArgumentException("Username or password provided was incorrect.  Check your configuration.");
+        }
+
+        return new DefaultMongoDBRecordWriter(db, collection);
 
     }
 
@@ -47,68 +51,5 @@ public void checkOutputSpecs(FileSystem fileSystem, JobConf jobConf) throws IOEx
         //Don't do anything with this.
     }
 
-    protected class MongoDBRecordWriter implements RecordWriter<K, V> {
-        private DB db;
-        private String collection;
-        private int retryAttempts;
-
-        protected MongoDBRecordWriter(DB db, String collection) {
-            this.db = db;
-            this.collection = collection;
-            this.retryAttempts = 10;
-        }
-
-        /**
-         * {@inheritDoc}
-         */
-        public synchronized void write(K k, V v) throws IOException {
-
-            log.debug("MongoDBRecordWriter: {key = " + k + ", value = " + v + ", collection = " + collection + "};");
-            BasicDBObject d = (BasicDBObject) v;
-
-            log.debug("Write Document: {document = " + d + ", collection = " + collection + "};");
-
-            int insertAttemptsRemaining = retryAttempts;
-
-            while(true)
-            {
-                try
-                {
-                    db.getCollection(collection).insert(d);
-                }
-                catch (MongoException e)
-                {
-                    if (insertAttemptsRemaining >= 0)
-                    {
-                        log.info("Waiting one second before retry...");
-                        try
-                        {
-                            Thread.sleep(1000);
-                        }
-                        catch (InterruptedException ie)
-                        {}
-                        log.info("Document insert failed: { message = " + e.getMessage() + ", attemptsRemaining = " + insertAttemptsRemaining);
-                        insertAttemptsRemaining--;
-                        continue;
-                    }
-                    else
-                    {
-                        throw new IOException("Exhausted 10 attempts to insert document: {document = " + d + "};", e);
-                    }
-                }
-                insertAttemptsRemaining = retryAttempts;
-                break;
-            }
-        }
-
-        /**
-         * {@inheritDoc}
-         */
-        public void close(Reporter reporter) throws IOException {
-        
-        }
-
-    }
-
 
 }

src/java/cascading/mongodb/MongoDBScheme.java

@@ -51,21 +51,21 @@ public Tuple source(Object key, Object value) {
     // {@inheritDoc}
     public void sink(TupleEntry tupleEntry, OutputCollector outputCollector) throws IOException {
 
-        BasicDBObject document = new BasicDBObject();
-
-        for (int i = 0; i < fields.length; i++) {
-            Fields field = fields[i];
-            TupleEntry values = tupleEntry.selectEntry(field);
-
-            for (int j = 0; j < values.getFields().size(); j++) {
-                Fields fields = values.getFields();
-                Tuple tuple = values.getTuple();
-
-                document.put(fields.get(j).toString(), tuple.getString(j));
-            }
-        }
-
-        outputCollector.collect(null, document);
+//        BasicDBObject document = new BasicDBObject();
+//
+//        for (int i = 0; i < fields.length; i++) {
+//            Fields field = fields[i];
+//            TupleEntry values = tupleEntry.selectEntry(field);
+//
+//            for (int j = 0; j < values.getFields().size(); j++) {
+//                Fields fields = values.getFields();
+//                Tuple tuple = values.getTuple();
+//
+//                document.put(fields.get(j).toString(), tuple.getString(j));
+//            }
+//        }
+
+        outputCollector.collect(null, tupleEntry);
 
 //        Tuple result = tupleEntry.selectTuple(getSinkFields());
 //        result = cleanTuple(result);

src/java/cascading/mongodb/MongoDBTap.java

@@ -103,11 +103,20 @@ private DBAddress getDBAddress()
 
     private DB getDB() {
 
+        log.debug("Requesting params for DB: {db=" + getMongo().getDB(getDBAddress().getDBName()) + "};");
         if (db == null) {
+            log.debug("DB was null - requesting refresh.");
             db = getMongo().getDB(getDBAddress().getDBName());
+
             if (username != null)
+            {
+                log.debug("Attempting to authenticate user...");
                 if (!db.authenticate(username, password))
+                {
                     throw new IllegalArgumentException("MongoDBTap: Auth Failed: {username = " + username + ", password = " + Arrays.toString(password) + "};");
+                }
+                log.debug("Apparently authentication passed for: {username=" + username + "};");
+            }
         }
 
         return db;
@@ -164,7 +173,7 @@ public boolean deletePath(JobConf jobConf) throws IOException {
             return true;
 
         log.debug("Removing collection: {name = " + collection + "};");
-        //getDB().getCollection(collection).drop();
+        getDB().getCollection(collection).drop();
 
         return true;
 

src/test/cascading/mongodb/MongoTest.java

@@ -1,13 +1,11 @@
 package cascading.mongodb;
 
-import com.mongodb.BasicDBObject;
-import com.mongodb.DB;
-import com.mongodb.DBCollection;
-import com.mongodb.Mongo;
+import com.mongodb.*;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
+import java.util.Arrays;
 import java.util.Set;
 
 /**
@@ -17,17 +15,21 @@
 public class MongoTest
 {
     private Mongo m;
+    private String username = "gameattain";
+    private char[] password = {'G', 'A', '.', '2', '0', '0', '9'};
+
+    //private DBAddress connection = new DBAddress("arrow.mongohq.com", 27025, "gameattain");
 
     @Before
     public void setup() throws Exception
     {
-       m = new Mongo("localhost"); 
+       m = new Mongo(new DBAddress("arrow.mongohq.com", 27025, "gameattain"));
     }
 
     @Test
     public void testMongoConnection() throws Exception
     {
-        Mongo pm = new Mongo("localhost");
+        Mongo pm = new Mongo(new DBAddress("arrow.mongohq.com", 27025, "gameattain"));
 
         Assert.assertNotNull(pm);
         System.out.println("Established connection with local Mongo instance.");
@@ -36,8 +38,13 @@ public void testMongoConnection() throws Exception
     @Test
     public void testGetCollection()
     {
-        DB db = m.getDB("testdb");
+        DB db = m.getDB("gameattain");
+
 
+        if (!db.authenticate(username, password))
+                {
+                    throw new IllegalArgumentException("MongoTest: Auth Failed: {username = " + username + ", password = " + Arrays.toString(password) + "};");
+                }
         Assert.assertNotNull(db);
 
         Set<String> collection = db.getCollectionNames();
@@ -54,7 +61,13 @@ public void exec(Object o)
     @Test
     public void testGetTestCollection()
     {
-        DB db = m.getDB("testdb");
+        DB db = m.getDB("gameattain");
+
+
+        if (!db.authenticate(username, password))
+                {
+                    throw new IllegalArgumentException("MongoTest: Auth Failed: {username = " + username + ", password = " + Arrays.toString(password) + "};");
+                }
 
         Assert.assertNotNull(db);
 
@@ -66,7 +79,13 @@ public void testGetTestCollection()
     @Test
     public void testInsertDoc()
     {
-        DB db = m.getDB("testdb");
+        DB db = m.getDB("gameattain");
+
+
+        if (!db.authenticate(username, password))
+                {
+                    throw new IllegalArgumentException("MongoTest: Auth Failed: {username = " + username + ", password = " + Arrays.toString(password) + "};");
+                }
         DBCollection collection = db.getCollection("foo");
 
         BasicDBObject doc = new BasicDBObject();