blob: e9172a32922f5d7cd3c2553309a2bedf456cd462 [file] [log] [blame]
/**
* 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.
*/
#include <cppunit/extensions/HelperMacros.h>
#include "CppAssertHelper.h"
#include <signal.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/select.h>
#include "CollectionUtil.h"
#include "ThreadingUtil.h"
using namespace Util;
#include "Vector.h"
using namespace std;
#include <cstring>
#include <list>
#include <zookeeper.h>
#include <errno.h>
#include <recordio.h>
#include "Util.h"
#ifdef THREADED
static void yield(zhandle_t *zh, int i)
{
sleep(i);
}
#else
static void yield(zhandle_t *zh, int seconds)
{
int fd;
int interest;
int events;
struct timeval tv;
int rc;
time_t expires = time(0) + seconds;
time_t timeLeft = seconds;
fd_set rfds, wfds, efds;
FD_ZERO(&rfds);
FD_ZERO(&wfds);
FD_ZERO(&efds);
while(timeLeft >= 0) {
zookeeper_interest(zh, &fd, &interest, &tv);
if (fd != -1) {
if (interest&ZOOKEEPER_READ) {
FD_SET(fd, &rfds);
} else {
FD_CLR(fd, &rfds);
}
if (interest&ZOOKEEPER_WRITE) {
FD_SET(fd, &wfds);
} else {
FD_CLR(fd, &wfds);
}
} else {
fd = 0;
}
FD_SET(0, &rfds);
if (tv.tv_sec > timeLeft) {
tv.tv_sec = timeLeft;
}
rc = select(fd+1, &rfds, &wfds, &efds, &tv);
timeLeft = expires - time(0);
events = 0;
if (FD_ISSET(fd, &rfds)) {
events |= ZOOKEEPER_READ;
}
if (FD_ISSET(fd, &wfds)) {
events |= ZOOKEEPER_WRITE;
}
zookeeper_process(zh, events);
}
}
#endif
typedef struct evt {
string path;
int type;
} evt_t;
typedef struct watchCtx {
private:
list<evt_t> events;
watchCtx(const watchCtx&);
watchCtx& operator=(const watchCtx&);
public:
bool connected;
zhandle_t *zh;
Mutex mutex;
watchCtx() {
connected = false;
zh = 0;
}
~watchCtx() {
if (zh) {
zookeeper_close(zh);
zh = 0;
}
}
evt_t getEvent() {
evt_t evt;
mutex.acquire();
CPPUNIT_ASSERT( events.size() > 0);
evt = events.front();
events.pop_front();
mutex.release();
return evt;
}
int countEvents() {
int count;
mutex.acquire();
count = events.size();
mutex.release();
return count;
}
void putEvent(evt_t evt) {
mutex.acquire();
events.push_back(evt);
mutex.release();
}
bool waitForConnected(zhandle_t *zh) {
time_t expires = time(0) + 10;
while(!connected && time(0) < expires) {
yield(zh, 1);
}
return connected;
}
bool waitForDisconnected(zhandle_t *zh) {
time_t expires = time(0) + 15;
while(connected && time(0) < expires) {
yield(zh, 1);
}
return !connected;
}
} watchctx_t;
class Zookeeper_multi : public CPPUNIT_NS::TestFixture
{
CPPUNIT_TEST_SUITE(Zookeeper_multi);
//FIXME: None of these tests pass in single-threaded mode. It seems to be a
//flaw in the test suite setup.
#ifdef THREADED
CPPUNIT_TEST(testCreate);
CPPUNIT_TEST(testCreateDelete);
CPPUNIT_TEST(testInvalidVersion);
CPPUNIT_TEST(testNestedCreate);
CPPUNIT_TEST(testSetData);
CPPUNIT_TEST(testUpdateConflict);
CPPUNIT_TEST(testDeleteUpdateConflict);
CPPUNIT_TEST(testAsyncMulti);
CPPUNIT_TEST(testMultiFail);
CPPUNIT_TEST(testCheck);
CPPUNIT_TEST(testWatch);
CPPUNIT_TEST(testSequentialNodeCreateInAsyncMulti);
CPPUNIT_TEST(testBigAsyncMulti);
#endif
CPPUNIT_TEST_SUITE_END();
static void watcher(zhandle_t *, int type, int state, const char *path,void*v){
watchctx_t *ctx = (watchctx_t*)v;
if (state == ZOO_CONNECTED_STATE) {
ctx->connected = true;
} else {
ctx->connected = false;
}
if (type != ZOO_SESSION_EVENT) {
evt_t evt;
evt.path = path;
evt.type = type;
ctx->putEvent(evt);
}
}
static const char hostPorts[];
const char *getHostPorts() {
return hostPorts;
}
zhandle_t *createClient(watchctx_t *ctx) {
return createClient(hostPorts, ctx);
}
zhandle_t *createClient(const char *hp, watchctx_t *ctx) {
zhandle_t *zk = zookeeper_init(hp, watcher, 10000, 0, ctx, 0);
ctx->zh = zk;
CPPUNIT_ASSERT_EQUAL(true, ctx->waitForConnected(zk));
return zk;
}
FILE *logfile;
public:
Zookeeper_multi() {
logfile = openlogfile("Zookeeper_multi");
}
~Zookeeper_multi() {
if (logfile) {
fflush(logfile);
fclose(logfile);
logfile = 0;
}
}
void setUp()
{
zoo_set_log_stream(logfile);
}
void tearDown()
{
}
static volatile int count;
static void multi_completion_fn(int rc, const void *data) {
CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
count++;
}
static void multi_completion_fn_no_assert(int rc, const void *data) {
count++;
}
static void multi_completion_fn_rc(int rc, const void *data) {
count++;
*((int*) data) = rc;
}
static void create_completion_fn_rc(int rc, const char* value, const void *data) {
count++;
*((int*) data) = rc;
}
static void waitForMultiCompletion(int seconds) {
time_t expires = time(0) + seconds;
while(count == 0 && time(0) < expires) {
sleep(1);
}
count--;
}
static void resetCounter() {
count = 0;
}
/**
* Test basic multi-op create functionality
*/
void testCreate() {
int rc;
watchctx_t ctx;
zhandle_t *zk = createClient(&ctx);
int sz = 512;
char p1[sz];
char p2[sz];
char p3[sz];
p1[0] = p2[0] = p3[0] = '\0';
int nops = 3 ;
zoo_op_t ops[nops];
zoo_op_result_t results[nops];
zoo_create_op_init(&ops[0], "/multi1", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
zoo_create_op_init(&ops[1], "/multi1/a", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p2, sz);
zoo_create_op_init(&ops[2], "/multi1/b", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p3, sz);
rc = zoo_multi(zk, nops, ops, results);
CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
CPPUNIT_ASSERT(strcmp(p1, "/multi1") == 0);
CPPUNIT_ASSERT(strcmp(p2, "/multi1/a") == 0);
CPPUNIT_ASSERT(strcmp(p3, "/multi1/b") == 0);
CPPUNIT_ASSERT_EQUAL((int)ZOK, results[0].err);
CPPUNIT_ASSERT_EQUAL((int)ZOK, results[1].err);
CPPUNIT_ASSERT_EQUAL((int)ZOK, results[2].err);
}
/**
* Test create followed by delete
*/
void testCreateDelete() {
int rc;
watchctx_t ctx;
zhandle_t *zk = createClient(&ctx);
int sz = 512;
char p1[sz];
p1[0] = '\0';
int nops = 2 ;
zoo_op_t ops[nops];
zoo_op_result_t results[nops];
zoo_create_op_init(&ops[0], "/multi2", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
zoo_delete_op_init(&ops[1], "/multi2", 0);
rc = zoo_multi(zk, nops, ops, results);
CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
// '/multi2' should have been deleted
rc = zoo_exists(zk, "/multi2", 0, NULL);
CPPUNIT_ASSERT_EQUAL((int)ZNONODE, rc);
}
/**
* Test invalid versions
*/
void testInvalidVersion() {
int rc;
watchctx_t ctx;
zhandle_t *zk = createClient(&ctx);
int nops = 4;
zoo_op_t ops[nops];
zoo_op_result_t results[nops];
zoo_create_op_init(&ops[0], "/multi3", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, NULL, 0);
zoo_delete_op_init(&ops[1], "/multi3", 1);
zoo_create_op_init(&ops[2], "/multi3", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, NULL, 0);
zoo_create_op_init(&ops[3], "/multi3/a", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, NULL, 0);
rc = zoo_multi(zk, nops, ops, results);
CPPUNIT_ASSERT_EQUAL((int)ZBADVERSION, rc);
CPPUNIT_ASSERT_EQUAL((int)ZOK, results[0].err);
CPPUNIT_ASSERT_EQUAL((int)ZBADVERSION, results[1].err);
CPPUNIT_ASSERT_EQUAL((int)ZRUNTIMEINCONSISTENCY, results[2].err);
CPPUNIT_ASSERT_EQUAL((int)ZRUNTIMEINCONSISTENCY, results[3].err);
}
/**
* Test nested creates that rely on state in earlier op in multi
*/
void testNestedCreate() {
int rc;
watchctx_t ctx;
zhandle_t *zk = createClient(&ctx);
int sz = 512;
char p1[sz];
p1[0] = '\0';
int nops = 6;
zoo_op_t ops[nops];
zoo_op_result_t results[nops];
/* Create */
zoo_create_op_init(&ops[0], "/multi4", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
zoo_create_op_init(&ops[1], "/multi4/a", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
zoo_create_op_init(&ops[2], "/multi4/a/1", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
/* Delete */
zoo_delete_op_init(&ops[3], "/multi4/a/1", 0);
zoo_delete_op_init(&ops[4], "/multi4/a", 0);
zoo_delete_op_init(&ops[5], "/multi4", 0);
rc = zoo_multi(zk, nops, ops, results);
CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
// Verify tree deleted
rc = zoo_exists(zk, "/multi4/a/1", 0, NULL);
CPPUNIT_ASSERT_EQUAL((int)ZNONODE, rc);
rc = zoo_exists(zk, "/multi4/a", 0, NULL);
CPPUNIT_ASSERT_EQUAL((int)ZNONODE, rc);
rc = zoo_exists(zk, "/multi4", 0, NULL);
CPPUNIT_ASSERT_EQUAL((int)ZNONODE, rc);
}
/**
* Test setdata functionality
*/
void testSetData() {
int rc;
watchctx_t ctx;
zhandle_t *zk = createClient(&ctx);
int sz = 512;
struct Stat s1;
char buf[sz];
int blen = sz ;
char p1[sz], p2[sz];
int nops = 2;
zoo_op_t ops[nops];
zoo_op_result_t results[nops];
zoo_create_op_init(&ops[0], "/multi5", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
zoo_create_op_init(&ops[1], "/multi5/a", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p2, sz);
rc = zoo_multi(zk, nops, ops, results);
CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
yield(zk, 5);
zoo_op_t setdata_ops[nops];
zoo_op_result_t setdata_results[nops];
zoo_set_op_init(&setdata_ops[0], "/multi5", "1", 1, 0, &s1);
zoo_set_op_init(&setdata_ops[1], "/multi5/a", "2", 1, 0, &s1);
rc = zoo_multi(zk, nops, setdata_ops, setdata_results);
CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
CPPUNIT_ASSERT_EQUAL((int)ZOK, results[0].err);
CPPUNIT_ASSERT_EQUAL((int)ZOK, results[1].err);
memset(buf, '\0', blen);
rc = zoo_get(zk, "/multi5", 0, buf, &blen, &s1);
CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
CPPUNIT_ASSERT_EQUAL(1, blen);
CPPUNIT_ASSERT(strcmp("1", buf) == 0);
memset(buf, '\0', blen);
rc = zoo_get(zk, "/multi5/a", 0, buf, &blen, &s1);
CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
CPPUNIT_ASSERT_EQUAL(1, blen);
CPPUNIT_ASSERT(strcmp("2", buf) == 0);
}
/**
* Test update conflicts
*/
void testUpdateConflict() {
int rc;
watchctx_t ctx;
zhandle_t *zk = createClient(&ctx);
int sz = 512;
char buf[sz];
int blen = sz;
char p1[sz];
p1[0] = '\0';
struct Stat s1;
int nops = 3;
zoo_op_t ops[nops];
zoo_op_result_t results[nops];
zoo_create_op_init(&ops[0], "/multi6", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
zoo_set_op_init(&ops[1], "/multi6", "X", 1, 0, &s1);
zoo_set_op_init(&ops[2], "/multi6", "Y", 1, 0, &s1);
rc = zoo_multi(zk, nops, ops, results);
CPPUNIT_ASSERT_EQUAL((int)ZBADVERSION, rc);
//Updating version solves conflict -- order matters
ops[2].set_op.version = 1;
rc = zoo_multi(zk, nops, ops, results);
CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
memset(buf, 0, sz);
rc = zoo_get(zk, "/multi6", 0, buf, &blen, &s1);
CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
CPPUNIT_ASSERT_EQUAL(blen, 1);
CPPUNIT_ASSERT(strncmp(buf, "Y", 1) == 0);
}
/**
* Test delete-update conflicts
*/
void testDeleteUpdateConflict() {
int rc;
watchctx_t ctx;
zhandle_t *zk = createClient(&ctx);
int sz = 512;
char buf[sz];
int blen;
char p1[sz];
p1[0] = '\0';
struct Stat stat;
int nops = 3;
zoo_op_t ops[nops];
zoo_op_result_t results[nops];
zoo_create_op_init(&ops[0], "/multi7", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
zoo_delete_op_init(&ops[1], "/multi7", 0);
zoo_set_op_init(&ops[2], "/multi7", "Y", 1, 0, &stat);
rc = zoo_multi(zk, nops, ops, results);
CPPUNIT_ASSERT_EQUAL((int)ZNONODE, rc);
// '/multi' should never have been created as entire op should fail
rc = zoo_exists(zk, "/multi7", 0, NULL);
CPPUNIT_ASSERT_EQUAL((int)ZNONODE, rc);
}
void testAsyncMulti() {
int rc;
watchctx_t ctx;
zhandle_t *zk = createClient(&ctx);
int sz = 512;
char p1[sz], p2[sz], p3[sz];
p1[0] = '\0';
p2[0] = '\0';
p3[0] = '\0';
int nops = 3;
zoo_op_t ops[nops];
zoo_op_result_t results[nops];
zoo_create_op_init(&ops[0], "/multi8", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
zoo_create_op_init(&ops[1], "/multi8/a", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p2, sz);
zoo_create_op_init(&ops[2], "/multi8/b", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p3, sz);
rc = zoo_amulti(zk, nops, ops, results, multi_completion_fn, 0);
waitForMultiCompletion(10);
CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
CPPUNIT_ASSERT(strcmp(p1, "/multi8") == 0);
CPPUNIT_ASSERT(strcmp(p2, "/multi8/a") == 0);
CPPUNIT_ASSERT(strcmp(p3, "/multi8/b") == 0);
CPPUNIT_ASSERT_EQUAL((int)ZOK, results[0].err);
CPPUNIT_ASSERT_EQUAL((int)ZOK, results[1].err);
CPPUNIT_ASSERT_EQUAL((int)ZOK, results[2].err);
}
void testMultiFail() {
int rc;
watchctx_t ctx;
zhandle_t *zk = createClient(&ctx);
int sz = 512;
char p1[sz], p2[sz], p3[sz];
p1[0] = '\0';
p2[0] = '\0';
p3[0] = '\0';
int nops = 3;
zoo_op_t ops[nops];
zoo_op_result_t results[nops];
zoo_create_op_init(&ops[0], "/multi9", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
zoo_create_op_init(&ops[1], "/multi9", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p2, sz);
zoo_create_op_init(&ops[2], "/multi9/b", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p3, sz);
rc = zoo_multi(zk, nops, ops, results);
CPPUNIT_ASSERT_EQUAL((int)ZNODEEXISTS, rc);
}
/**
* Test basic multi-op check functionality
*/
void testCheck() {
int rc;
watchctx_t ctx;
zhandle_t *zk = createClient(&ctx);
int sz = 512;
char p1[sz];
p1[0] = '\0';
struct Stat s1;
rc = zoo_create(zk, "/multi0", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
// Conditionally create /multi0/a' only if '/multi0' at version 0
int nops = 2;
zoo_op_t ops[nops];
zoo_op_result_t results[nops];
zoo_check_op_init(&ops[0], "/multi0", 0);
zoo_create_op_init(&ops[1], "/multi0/a", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
rc = zoo_multi(zk, nops, ops, results);
CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
CPPUNIT_ASSERT_EQUAL((int)ZOK, results[0].err);
CPPUNIT_ASSERT_EQUAL((int)ZOK, results[1].err);
// '/multi0/a' should have been created as it passed version check
rc = zoo_exists(zk, "/multi0/a", 0, NULL);
CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
// Only create '/multi0/b' if '/multi0' at version 10 (which it's not)
zoo_op_t ops2[nops];
zoo_check_op_init(&ops2[0], "/multi0", 10);
zoo_create_op_init(&ops2[1], "/multi0/b", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, p1, sz);
rc = zoo_multi(zk, nops, ops2, results);
CPPUNIT_ASSERT_EQUAL((int)ZBADVERSION, rc);
CPPUNIT_ASSERT_EQUAL((int)ZBADVERSION, results[0].err);
CPPUNIT_ASSERT_EQUAL((int)ZRUNTIMEINCONSISTENCY, results[1].err);
// '/multi0/b' should NOT have been created
rc = zoo_exists(zk, "/multi0/b", 0, NULL);
CPPUNIT_ASSERT_EQUAL((int)ZNONODE, rc);
}
/**
* Do a multi op inside a watch callback context.
*/
static void doMultiInWatch(zhandle_t *zk, int type, int state, const char *path, void *ctx) {
int rc;
int sz = 512;
char p1[sz];
p1[0] = '\0';
struct Stat s1;
int nops = 1;
zoo_op_t ops[nops];
zoo_op_result_t results[nops];
zoo_set_op_init(&ops[0], "/multiwatch", "1", 1, -1, NULL);
rc = zoo_multi(zk, nops, ops, results);
CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
CPPUNIT_ASSERT_EQUAL((int)ZOK, results[0].err);
memset(p1, '\0', sz);
rc = zoo_get(zk, "/multiwatch", 0, p1, &sz, &s1);
CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
CPPUNIT_ASSERT_EQUAL(1, sz);
CPPUNIT_ASSERT(strcmp("1", p1) == 0);
count++;
}
/**
* Test multi-op called from a watch
*/
void testWatch() {
int rc;
watchctx_t ctx;
zhandle_t *zk = createClient(&ctx);
int sz = 512;
char p1[sz];
p1[0] = '\0';
rc = zoo_create(zk, "/multiwatch", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, NULL, 0);
CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
// create a watch on node '/multiwatch'
rc = zoo_wget(zk, "/multiwatch", doMultiInWatch, &ctx, p1, &sz, NULL);
// setdata on node '/multiwatch' this should trip the watch
rc = zoo_set(zk, "/multiwatch", NULL, -1, -1);
CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
// wait for multi completion in doMultiInWatch
waitForMultiCompletion(5);
}
/**
* ZOOKEEPER-1636: If request is too large, the server will cut the
* connection without sending response packet. The client will try to
* process completion on multi request and eventually cause SIGSEGV
*/
void testBigAsyncMulti() {
int rc;
int callback_rc = (int) ZOK;
watchctx_t ctx;
zhandle_t *zk = createClient(&ctx);
// The request should be more than 1MB which exceeds the default
// jute.maxbuffer and causes the server to drop client connection
const int iteration = 500;
const int type_count = 3;
const int nops = iteration * type_count;
char buff[1024];
zoo_op_result_t results[nops];
zoo_op_t ops[nops];
struct Stat* s[nops];
int index = 0;
// Test that we deliver error to 3 types of sub-request
for (int i = 0; i < iteration; ++i) {
zoo_set_op_init(&ops[index++], "/x", buff, sizeof(buff), -1, s[i]);
zoo_create_op_init(&ops[index++], "/x", buff, sizeof(buff),
&ZOO_OPEN_ACL_UNSAFE, ZOO_SEQUENCE, NULL, 0);
zoo_delete_op_init(&ops[index++], "/x", -1);
}
rc = zoo_amulti(zk, nops, ops, results, multi_completion_fn_rc,
&callback_rc);
CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
waitForMultiCompletion(10);
// With the bug, we will get SIGSEGV before reaching this point
CPPUNIT_ASSERT_EQUAL((int) ZCONNECTIONLOSS, callback_rc);
// Make sure that all sub-request completions get processed
for (int i = 0; i < nops; ++i) {
CPPUNIT_ASSERT_EQUAL((int) ZCONNECTIONLOSS, results[i].err);
}
// The handle should be able to recover itself.
ctx.waitForConnected(zk);
// Try to submit another async request to see if it get processed
// correctly
rc = zoo_acreate(zk, "/target", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0,
create_completion_fn_rc, &callback_rc);
CPPUNIT_ASSERT_EQUAL((int) ZOK, rc);
waitForMultiCompletion(10);
CPPUNIT_ASSERT_EQUAL((int) ZOK, callback_rc);
}
/**
* ZOOKEEPER-1624: PendingChanges of create sequential node request didn't
* get rollbacked correctly when multi-op failed. This caused
* create sequential node request in subsequent multi-op to failed because
* sequential node name generation is incorrect.
*
* The check is to make sure that each request in multi-op failed with
* the correct reason.
*/
void testSequentialNodeCreateInAsyncMulti() {
int rc;
watchctx_t ctx;
zhandle_t *zk = createClient(&ctx);
int iteration = 4;
int nops = 2;
zoo_op_result_t results[iteration][nops];
zoo_op_t ops[nops];
zoo_create_op_init(&ops[0], "/node-", "", 0, &ZOO_OPEN_ACL_UNSAFE, ZOO_SEQUENCE, NULL, 0);
zoo_create_op_init(&ops[1], "/dup", "", 0, &ZOO_OPEN_ACL_UNSAFE, 0, NULL, 0);
for (int i = 0; i < iteration ; ++i) {
rc = zoo_amulti(zk, nops, ops, results[i], multi_completion_fn_no_assert, 0);
CPPUNIT_ASSERT_EQUAL((int)ZOK, rc);
}
waitForMultiCompletion(10);
CPPUNIT_ASSERT_EQUAL((int)ZOK, results[0][0].err);
CPPUNIT_ASSERT_EQUAL((int)ZOK, results[1][0].err);
CPPUNIT_ASSERT_EQUAL((int)ZOK, results[2][0].err);
CPPUNIT_ASSERT_EQUAL((int)ZOK, results[3][0].err);
CPPUNIT_ASSERT_EQUAL((int)ZOK, results[0][1].err);
CPPUNIT_ASSERT_EQUAL((int)ZNODEEXISTS, results[1][1].err);
CPPUNIT_ASSERT_EQUAL((int)ZNODEEXISTS, results[2][1].err);
CPPUNIT_ASSERT_EQUAL((int)ZNODEEXISTS, results[3][1].err);
resetCounter();
}
};
volatile int Zookeeper_multi::count;
const char Zookeeper_multi::hostPorts[] = "127.0.0.1:22181";
CPPUNIT_TEST_SUITE_REGISTRATION(Zookeeper_multi);