HDFS-9093. Initialize protobuf fields in RemoteBlockReaderTest. Contributed by Haohui Mai.

hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/tests/mock_connection.cc

@@ -20,6 +20,10 @@
 
 namespace hdfs {
 
+MockConnectionBase::MockConnectionBase(::asio::io_service *io_service)
+    : io_service_(io_service)
+{}
+
 MockConnectionBase::~MockConnectionBase() {}
 
 }

hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/tests/mock_connection.h

@@ -21,12 +21,15 @@
 #include <asio/error_code.hpp>
 #include <asio/buffer.hpp>
 #include <asio/streambuf.hpp>
+#include <asio/io_service.hpp>
+
 #include <gmock/gmock.h>
 
 namespace hdfs {
 
 class MockConnectionBase {
 public:
+  MockConnectionBase(::asio::io_service *io_service);
   virtual ~MockConnectionBase();
   typedef std::pair<asio::error_code, std::string> ProducerResult;
   template <class MutableBufferSequence, class Handler>
@@ -34,7 +37,7 @@ public:
     if (produced_.size() == 0) {
       ProducerResult r = Produce();
       if (r.first) {
-        handler(r.first, 0);
+        io_service_->post(std::bind(handler, r.first, 0));
       }
       asio::mutable_buffers_1 data = produced_.prepare(r.second.size());
       asio::buffer_copy(data, asio::buffer(r.second));
@@ -44,17 +47,18 @@ public:
     size_t len = std::min(asio::buffer_size(buf), produced_.size());
     asio::buffer_copy(buf, produced_.data());
     produced_.consume(len);
-    handler(asio::error_code(), len);
+    io_service_->post(std::bind(handler, asio::error_code(), len));
   }
 
   template <class ConstBufferSequence, class Handler>
   void async_write_some(const ConstBufferSequence &buf, Handler &&handler) {
     // CompletionResult res = OnWrite(buf);
-    handler(asio::error_code(), asio::buffer_size(buf));
+    io_service_->post(std::bind(handler, asio::error_code(), asio::buffer_size(buf)));
   }
 
 protected:
   virtual ProducerResult Produce() = 0;
+  ::asio::io_service *io_service_;
 
 private:
   asio::streambuf produced_;

hadoop-hdfs-project/hadoop-hdfs-client/src/main/native/libhdfspp/tests/remote_block_reader_test.cc

@@ -51,6 +51,8 @@ namespace hdfs {
 
 class MockDNConnection : public MockConnectionBase {
 public:
+  MockDNConnection(::asio::io_service &io_service)
+      : MockConnectionBase(&io_service) {}
   MOCK_METHOD0(Produce, ProducerResult());
 };
 }
@@ -91,35 +93,30 @@ ProducePacket(const std::string &data, const std::string &checksum,
   return std::make_pair(error_code(), std::move(payload));
 }
 
-template<class Stream = MockDNConnection>
+template <class Stream = MockDNConnection, class Handler>
 static std::shared_ptr<RemoteBlockReader<Stream>>
-ReadContent(Stream *conn, TokenProto *token,
-            const ExtendedBlockProto &block, uint64_t length, uint64_t offset,
-            const mutable_buffers_1 &buf, Status *status, size_t *transferred) {
+ReadContent(Stream *conn, TokenProto *token, const ExtendedBlockProto &block,
+            uint64_t length, uint64_t offset, const mutable_buffers_1 &buf,
+            const Handler &handler) {
   BlockReaderOptions options;
-  auto reader =
-      std::make_shared<RemoteBlockReader<Stream>>(options, conn);
+  auto reader = std::make_shared<RemoteBlockReader<Stream>>(options, conn);
   Status result;
-  reader->async_connect(
-      "libhdfs++", token, &block, length, offset,
-      [buf, reader, status, transferred](const Status &stat) {
-        if (!stat.ok()) {
-          *status = stat;
-        } else {
-          reader->async_read_some(
-              buf, [status, transferred](const Status &stat, size_t t) {
-                *transferred = t;
-                *status = stat;
-              });
-        }
-      });
+  reader->async_connect("libhdfs++", token, &block, length, offset,
+                        [buf, reader, handler](const Status &stat) {
+                          if (!stat.ok()) {
+                            handler(stat, 0);
+                          } else {
+                            reader->async_read_some(buf, handler);
+                          }
+                        });
   return reader;
 }
 
 TEST(RemoteBlockReaderTest, TestReadWholeBlock) {
   static const size_t kChunkSize = 512;
   static const string kChunkData(kChunkSize, 'a');
-  MockDNConnection conn;
+  ::asio::io_service io_service;
+  MockDNConnection conn(io_service);
   BlockOpResponseProto block_op_resp;
 
   block_op_resp.set_status(::hadoop::hdfs::Status::SUCCESS);
@@ -128,15 +125,20 @@ TEST(RemoteBlockReaderTest, TestReadWholeBlock) {
       .WillOnce(Return(ProducePacket(kChunkData, "", 0, 1, true)));
 
   ExtendedBlockProto block;
+  block.set_poolid("foo");
+  block.set_blockid(0);
+  block.set_generationstamp(0);
+  
   std::string data(kChunkSize, 0);
-  size_t transferred = 0;
-  Status stat;
   ReadContent(&conn, nullptr, block, kChunkSize, 0,
-              buffer(const_cast<char *>(data.c_str()), data.size()), &stat,
-              &transferred);
-  ASSERT_TRUE(stat.ok());
-  ASSERT_EQ(kChunkSize, transferred);
-  ASSERT_EQ(kChunkData, data);
+              buffer(const_cast<char *>(data.c_str()), data.size()),
+              [&data, &io_service](const Status &stat, size_t transferred) {
+                ASSERT_TRUE(stat.ok());
+                ASSERT_EQ(kChunkSize, transferred);
+                ASSERT_EQ(kChunkData, data);
+                io_service.stop();
+              });
+  io_service.run();
 }
 
 TEST(RemoteBlockReaderTest, TestReadWithinChunk) {
@@ -145,7 +147,8 @@ TEST(RemoteBlockReaderTest, TestReadWithinChunk) {
   static const size_t kOffset = kChunkSize / 4;
   static const string kChunkData = string(kOffset, 'a') + string(kLength, 'b');
 
-  MockDNConnection conn;
+  ::asio::io_service io_service;
+  MockDNConnection conn(io_service);
   BlockOpResponseProto block_op_resp;
   ReadOpChecksumInfoProto *checksum_info =
       block_op_resp.mutable_readopchecksuminfo();
@@ -160,22 +163,28 @@ TEST(RemoteBlockReaderTest, TestReadWithinChunk) {
       .WillOnce(Return(ProducePacket(kChunkData, "", kOffset, 1, true)));
 
   ExtendedBlockProto block;
+  block.set_poolid("foo");
+  block.set_blockid(0);
+  block.set_generationstamp(0);
+
   string data(kLength, 0);
-  size_t transferred = 0;
-  Status stat;
   ReadContent(&conn, nullptr, block, data.size(), kOffset,
-              buffer(const_cast<char *>(data.c_str()), data.size()), &stat,
-              &transferred);
-  ASSERT_TRUE(stat.ok());
-  ASSERT_EQ(kLength, transferred);
-  ASSERT_EQ(kChunkData.substr(kOffset, kLength), data);
+              buffer(const_cast<char *>(data.c_str()), data.size()),
+              [&data, &io_service](const Status &stat, size_t transferred) {
+                ASSERT_TRUE(stat.ok());
+                ASSERT_EQ(kLength, transferred);
+                ASSERT_EQ(kChunkData.substr(kOffset, kLength), data);
+                io_service.stop();
+              });
+  io_service.run();
 }
 
 TEST(RemoteBlockReaderTest, TestReadMultiplePacket) {
   static const size_t kChunkSize = 1024;
   static const string kChunkData(kChunkSize, 'a');
 
-  MockDNConnection conn;
+  ::asio::io_service io_service;
+  MockDNConnection conn(io_service);
   BlockOpResponseProto block_op_resp;
   block_op_resp.set_status(::hadoop::hdfs::Status::SUCCESS);
 
@@ -185,25 +194,37 @@ TEST(RemoteBlockReaderTest, TestReadMultiplePacket) {
       .WillOnce(Return(ProducePacket(kChunkData, "", kChunkSize, 2, true)));
 
   ExtendedBlockProto block;
+  block.set_poolid("foo");
+  block.set_blockid(0);
+  block.set_generationstamp(0);
+
   string data(kChunkSize, 0);
-  size_t transferred = 0;
-  Status stat;
   mutable_buffers_1 buf = buffer(const_cast<char *>(data.c_str()), data.size());
-  auto reader = ReadContent(&conn, nullptr, block, data.size(), 0, buf, &stat,
-                            &transferred);
-  ASSERT_TRUE(stat.ok());
-  ASSERT_EQ(kChunkSize, transferred);
-  ASSERT_EQ(kChunkData, data);
-
-  data.clear();
-  data.resize(kChunkSize);
-  transferred = 0;
-
-  reader->async_read_some(buf, [&data](const Status &stat, size_t transferred) {
-    ASSERT_TRUE(stat.ok());
-    ASSERT_EQ(kChunkSize, transferred);
-    ASSERT_EQ(kChunkData, data);
-  });
+  BlockReaderOptions options;
+  auto reader = std::make_shared<RemoteBlockReader<MockDNConnection> >(options, &conn);
+  Status result;
+  reader->async_connect(
+      "libhdfs++", nullptr, &block, data.size(), 0,
+      [buf, reader, &data, &io_service](const Status &stat) {
+        ASSERT_TRUE(stat.ok());
+        reader->async_read_some(
+            buf, [buf, reader, &data, &io_service](const Status &stat, size_t transferred) {
+              ASSERT_TRUE(stat.ok());
+              ASSERT_EQ(kChunkSize, transferred);
+              ASSERT_EQ(kChunkData, data);
+              data.clear();
+              data.resize(kChunkSize);
+              transferred = 0;
+              reader->async_read_some(
+                  buf, [&data,&io_service](const Status &stat, size_t transferred) {
+                    ASSERT_TRUE(stat.ok());
+                    ASSERT_EQ(kChunkSize, transferred);
+                    ASSERT_EQ(kChunkData, data);
+                    io_service.stop();
+                  });
+            });
+      });
+  io_service.run();
 }
 
 TEST(RemoteBlockReaderTest, TestSaslConnection) {
@@ -212,7 +233,8 @@ TEST(RemoteBlockReaderTest, TestSaslConnection) {
   static const string kAuthPayload = "realm=\"0\",nonce=\"+GAWc+O6yEAWpew/"
                                      "qKah8qh4QZLoOLCDcTtEKhlS\",qop=\"auth\","
                                      "charset=utf-8,algorithm=md5-sess";
-  MockDNConnection conn;
+  ::asio::io_service io_service;
+  MockDNConnection conn(io_service);
   BlockOpResponseProto block_op_resp;
   block_op_resp.set_status(::hadoop::hdfs::Status::SUCCESS);
 
@@ -233,20 +255,24 @@ TEST(RemoteBlockReaderTest, TestSaslConnection) {
 
   DataTransferSaslStream<MockDNConnection> sasl_conn(&conn, "foo", "bar");
   ExtendedBlockProto block;
+  block.set_poolid("foo");
+  block.set_blockid(0);
+  block.set_generationstamp(0);
+
   std::string data(kChunkSize, 0);
-  size_t transferred = 0;
-  Status stat;
-  sasl_conn.Handshake([&stat](const Status &s) {
-      stat = s;
-    });
-
-  ASSERT_TRUE(stat.ok());
-  ReadContent(&sasl_conn, nullptr, block, kChunkSize, 0,
-              buffer(const_cast<char *>(data.c_str()), data.size()), &stat,
-              &transferred);
-  ASSERT_TRUE(stat.ok());
-  ASSERT_EQ(kChunkSize, transferred);
-  ASSERT_EQ(kChunkData, data);
+  sasl_conn.Handshake([&sasl_conn, &block, &data, &io_service](
+      const Status &s) {
+    ASSERT_TRUE(s.ok());
+    ReadContent(&sasl_conn, nullptr, block, kChunkSize, 0,
+                buffer(const_cast<char *>(data.c_str()), data.size()),
+                [&data, &io_service](const Status &stat, size_t transferred) {
+                  ASSERT_TRUE(stat.ok());
+                  ASSERT_EQ(kChunkSize, transferred);
+                  ASSERT_EQ(kChunkData, data);
+                  io_service.stop();
+                });
+  });
+  io_service.run();
 }
 
 int main(int argc, char *argv[]) {