kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jun...@apache.org
Subject git commit: kafka-791; Fix validation bugs in System Test; patched by John Fung; reviewed by Jun Rao
Date Mon, 25 Mar 2013 21:08:45 GMT
Updated Branches:
  refs/heads/0.8 e367f3ffb -> 26c50fac4


kafka-791; Fix validation bugs in System Test; patched by John Fung; reviewed by Jun Rao


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/26c50fac
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/26c50fac
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/26c50fac

Branch: refs/heads/0.8
Commit: 26c50fac47802e4aa03793c60cb8316995bb37f1
Parents: e367f3f
Author: John Fung <fung.john@gmail.com>
Authored: Mon Mar 25 14:08:24 2013 -0700
Committer: Jun Rao <junrao@gmail.com>
Committed: Mon Mar 25 14:08:24 2013 -0700

----------------------------------------------------------------------
 .../mirror_maker_testsuite/mirror_maker_test.py    |    7 +-
 .../replication_testsuite/replica_basic_test.py    |   15 +-
 system_test/utils/kafka_system_test_utils.py       |  304 ++++++--------
 system_test/utils/replication_utils.py             |    3 +
 system_test/utils/system_test_utils.py             |   78 ++++-
 5 files changed, 222 insertions(+), 185 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/26c50fac/system_test/mirror_maker_testsuite/mirror_maker_test.py
----------------------------------------------------------------------
diff --git a/system_test/mirror_maker_testsuite/mirror_maker_test.py b/system_test/mirror_maker_testsuite/mirror_maker_test.py
index 48b0d25..098f531 100644
--- a/system_test/mirror_maker_testsuite/mirror_maker_test.py
+++ b/system_test/mirror_maker_testsuite/mirror_maker_test.py
@@ -76,6 +76,8 @@ class MirrorMakerTest(ReplicationUtils, SetupUtils):
             self.testSuiteAbsPathName, SystemTestEnv.SYSTEM_TEST_CASE_PREFIX)
         testCasePathNameList.sort()
 
+        replicationUtils = ReplicationUtils(self)
+
         # =============================================================
         # launch each testcase one by one: testcase_1, testcase_2, ...
         # =============================================================
@@ -282,9 +284,8 @@ class MirrorMakerTest(ReplicationUtils, SetupUtils):
                 # validate the data matched and checksum
                 # =============================================
                 self.log_message("validating data matched")
-                #kafka_system_test_utils.validate_data_matched(self.systemTestEnv, self.testcaseEnv)
-                kafka_system_test_utils.validate_simple_consumer_data_matched(self.systemTestEnv,
self.testcaseEnv)
-                kafka_system_test_utils.validate_broker_log_segment_checksum(self.systemTestEnv,
self.testcaseEnv)
+                kafka_system_test_utils.validate_data_matched(self.systemTestEnv, self.testcaseEnv,
replicationUtils)
+                kafka_system_test_utils.validate_broker_log_segment_checksum(self.systemTestEnv,
self.testcaseEnv, "source")
                 kafka_system_test_utils.validate_broker_log_segment_checksum(self.systemTestEnv,
self.testcaseEnv, "target")
 
                 # =============================================

http://git-wip-us.apache.org/repos/asf/kafka/blob/26c50fac/system_test/replication_testsuite/replica_basic_test.py
----------------------------------------------------------------------
diff --git a/system_test/replication_testsuite/replica_basic_test.py b/system_test/replication_testsuite/replica_basic_test.py
index 3fc47d9..ce29240 100644
--- a/system_test/replication_testsuite/replica_basic_test.py
+++ b/system_test/replication_testsuite/replica_basic_test.py
@@ -77,6 +77,8 @@ class ReplicaBasicTest(ReplicationUtils, SetupUtils):
             self.testSuiteAbsPathName, SystemTestEnv.SYSTEM_TEST_CASE_PREFIX)
         testCasePathNameList.sort()
 
+        replicationUtils = ReplicationUtils(self)
+
         # =============================================================
         # launch each testcase one by one: testcase_1, testcase_2, ...
         # =============================================================
@@ -423,16 +425,15 @@ class ReplicaBasicTest(ReplicationUtils, SetupUtils):
                 self.log_message("validating data matched")
 
                 if logRetentionTest.lower() == "true":
-                    kafka_system_test_utils.validate_simple_consumer_data_matched_across_replicas(self.systemTestEnv,
self.testcaseEnv)
-                    kafka_system_test_utils.validate_data_matched(self.systemTestEnv, self.testcaseEnv)
+                    kafka_system_test_utils.validate_data_matched(self.systemTestEnv, self.testcaseEnv,
replicationUtils)
                 elif consumerMultiTopicsMode.lower() == "true":
-                    #kafka_system_test_utils.validate_broker_log_segment_checksum(self.systemTestEnv,
self.testcaseEnv)
-                    kafka_system_test_utils.validate_data_matched_in_multi_topics_from_single_consumer_producer(self.systemTestEnv,
self.testcaseEnv)
+                    kafka_system_test_utils.validate_data_matched_in_multi_topics_from_single_consumer_producer(
+                        self.systemTestEnv, self.testcaseEnv, replicationUtils)
                 else:
                     kafka_system_test_utils.validate_simple_consumer_data_matched_across_replicas(self.systemTestEnv,
self.testcaseEnv)
-                    #kafka_system_test_utils.validate_broker_log_segment_checksum(self.systemTestEnv,
self.testcaseEnv)
-                    kafka_system_test_utils.validate_data_matched(self.systemTestEnv, self.testcaseEnv)
-
+                    kafka_system_test_utils.validate_broker_log_segment_checksum(self.systemTestEnv,
self.testcaseEnv)
+                    kafka_system_test_utils.validate_data_matched(self.systemTestEnv, self.testcaseEnv,
replicationUtils)
+ 
                 # =============================================
                 # draw graphs
                 # =============================================

http://git-wip-us.apache.org/repos/asf/kafka/blob/26c50fac/system_test/utils/kafka_system_test_utils.py
----------------------------------------------------------------------
diff --git a/system_test/utils/kafka_system_test_utils.py b/system_test/utils/kafka_system_test_utils.py
index 9411405..9e58624 100644
--- a/system_test/utils/kafka_system_test_utils.py
+++ b/system_test/utils/kafka_system_test_utils.py
@@ -1189,17 +1189,21 @@ def get_message_checksum(logPathName):
     return messageChecksumList
 
 
-def validate_data_matched(systemTestEnv, testcaseEnv):
+def validate_data_matched(systemTestEnv, testcaseEnv, replicationUtils):
+    logger.debug("#### Inside validate_data_matched", extra=d)
+
     validationStatusDict        = testcaseEnv.validationStatusDict
     clusterEntityConfigDictList = systemTestEnv.clusterEntityConfigDictList
 
     prodPerfCfgList = system_test_utils.get_dict_from_list_of_dicts(clusterEntityConfigDictList,
"role", "producer_performance")
     consumerCfgList = system_test_utils.get_dict_from_list_of_dicts(clusterEntityConfigDictList,
"role", "console_consumer")
 
+    consumerDuplicateCount = 0
+
     for prodPerfCfg in prodPerfCfgList:
         producerEntityId = prodPerfCfg["entity_id"]
-        #topic = system_test_utils.get_data_by_lookup_keyval(testcaseEnv.testcaseConfigsList,
"entity_id", producerEntityId, "topic")
-        topic = testcaseEnv.producerTopicsString
+        topic = system_test_utils.get_data_by_lookup_keyval(testcaseEnv.testcaseConfigsList,
"entity_id", producerEntityId, "topic")
+        logger.debug("working on topic : " + topic, extra=d)
         acks  = system_test_utils.get_data_by_lookup_keyval(testcaseEnv.testcaseConfigsList,
"entity_id", producerEntityId, "request-num-acks")
 
         consumerEntityIdList = system_test_utils.get_data_from_list_of_dicts( \
@@ -1207,13 +1211,14 @@ def validate_data_matched(systemTestEnv, testcaseEnv):
 
         matchingConsumerEntityId = None
         for consumerEntityId in consumerEntityIdList:
-            #consumerTopic = system_test_utils.get_data_by_lookup_keyval(testcaseEnv.testcaseConfigsList,
"entity_id", consumerEntityId, "topic")
-            consumerTopic = testcaseEnv.consumerTopicsString
+            consumerTopic = system_test_utils.get_data_by_lookup_keyval(testcaseEnv.testcaseConfigsList,
"entity_id", consumerEntityId, "topic")
             if consumerTopic in topic:
                 matchingConsumerEntityId = consumerEntityId
+                logger.debug("matching consumer entity id found", extra=d)
                 break
 
         if matchingConsumerEntityId is None:
+            logger.debug("matching consumer entity id NOT found", extra=d)
             break
 
         msgIdMissingInConsumerLogPathName = get_testcase_config_log_dir_pathname( \
@@ -1229,10 +1234,11 @@ def validate_data_matched(systemTestEnv, testcaseEnv):
         producerMsgIdSet   = set(producerMsgIdList)
         consumerMsgIdSet   = set(consumerMsgIdList)
 
-        missingMsgIdInConsumer = producerMsgIdSet - consumerMsgIdSet
+        consumerDuplicateCount = len(consumerMsgIdList) - len(consumerMsgIdSet)
+        missingUniqConsumerMsgId = system_test_utils.subtract_list(producerMsgIdSet, consumerMsgIdSet)
 
         outfile = open(msgIdMissingInConsumerLogPathName, "w")
-        for id in missingMsgIdInConsumer:
+        for id in missingUniqConsumerMsgId:
             outfile.write(id + "\n")
         outfile.close()
 
@@ -1241,20 +1247,28 @@ def validate_data_matched(systemTestEnv, testcaseEnv):
         validationStatusDict["Unique messages from producer on [" + topic + "]"] = str(len(producerMsgIdSet))
         validationStatusDict["Unique messages from consumer on [" + topic + "]"] = str(len(consumerMsgIdSet))
 
-        if ( len(missingMsgIdInConsumer) == 0 and len(producerMsgIdSet) > 0 ):
+        missingPercentage = len(missingUniqConsumerMsgId) * 100.00 / len(producerMsgIdSet)
+        logger.info("Data loss threshold % : " + str(replicationUtils.ackOneDataLossThresholdPercent),
extra=d)
+        logger.warn("Data loss % on topic : " + topic + " : " + str(missingPercentage), extra=d)
+
+        if ( len(missingUniqConsumerMsgId) == 0 and len(producerMsgIdSet) > 0 ):
             validationStatusDict["Validate for data matched on topic [" + topic + "]"] =
"PASSED"
         elif (acks == "1"):
-            missingPercentage = len(missingMsgIdInConsumer) * 100 / len(producerMsgIdSet)
-            print "#### missing Percent : ", missingPercentage
-            if missingPercentage <= 1:
+            if missingPercentage <= replicationUtils.ackOneDataLossThresholdPercent:
                 validationStatusDict["Validate for data matched on topic [" + topic + "]"]
= "PASSED"
-                logger.warn("Test case passes with less than 1% data loss : [" + str(len(missingMsgIdInConsumer))
+ "] missing messages", extra=d)
+                logger.warn("Test case (Acks = 1) passes with less than " + str(replicationUtils.ackOneDataLossThresholdPercent)
\
+                    + "% data loss : [" + str(len(missingUniqConsumerMsgId)) + "] missing
messages", extra=d)
+            else:
+                validationStatusDict["Validate for data matched on topic [" + topic + "]"]
= "FAILED"
+                logger.error("Test case (Acks = 1) failed with more than " + str(replicationUtils.ackOneDataLossThresholdPercent)
\
+                    + "% data loss : [" + str(len(missingUniqConsumerMsgId)) + "] missing
messages", extra=d)
         else:
             validationStatusDict["Validate for data matched on topic [" + topic + "]"] =
"FAILED"
             logger.info("See " + msgIdMissingInConsumerLogPathName + " for missing MessageID",
extra=d)
 
 
 def validate_leader_election_successful(testcaseEnv, leaderDict, validationStatusDict):
+    logger.debug("#### Inside validate_leader_election_successful", extra=d)
 
     if ( len(leaderDict) > 0 ):
         try:
@@ -1545,6 +1559,8 @@ def start_migration_tool(systemTestEnv, testcaseEnv, onlyThisEntityId=None):
 
 
 def validate_07_08_migrated_data_matched(systemTestEnv, testcaseEnv):
+    logger.debug("#### Inside validate_07_08_migrated_data_matched", extra=d)
+
     validationStatusDict        = testcaseEnv.validationStatusDict
     clusterEntityConfigDictList = systemTestEnv.clusterEntityConfigDictList
 
@@ -1614,6 +1630,7 @@ def validate_07_08_migrated_data_matched(systemTestEnv, testcaseEnv):
             logger.info("See " + msgChecksumMissingInConsumerLogPathName + " for missing
MessageID", extra=d)
 
 def validate_broker_log_segment_checksum(systemTestEnv, testcaseEnv, clusterName="source"):
+    logger.debug("#### Inside validate_broker_log_segment_checksum", extra=d)
 
     anonLogger.info("================================================")
     anonLogger.info("validating merged broker log segment checksums")
@@ -1823,77 +1840,6 @@ def start_simple_consumer(systemTestEnv, testcaseEnv, minStartingOffsetDict=None
                 partitionId += 1
             replicaIndex += 1
 
-def validate_simple_consumer_data_matched(systemTestEnv, testcaseEnv):
-    validationStatusDict        = testcaseEnv.validationStatusDict
-    clusterEntityConfigDictList = systemTestEnv.clusterEntityConfigDictList
-
-    prodPerfCfgList = system_test_utils.get_dict_from_list_of_dicts(clusterEntityConfigDictList,
"role", "producer_performance")
-    consumerCfgList = system_test_utils.get_dict_from_list_of_dicts(clusterEntityConfigDictList,
"role", "console_consumer")
-
-    mismatchCount = 0
-
-    for prodPerfCfg in prodPerfCfgList:
-        producerEntityId = prodPerfCfg["entity_id"]
-        topic = system_test_utils.get_data_by_lookup_keyval(testcaseEnv.testcaseConfigsList,
"entity_id", producerEntityId, "topic")
-        acks  = system_test_utils.get_data_by_lookup_keyval(testcaseEnv.testcaseConfigsList,
"entity_id", producerEntityId, "request-num-acks")
-        logger.debug("request-num-acks [" + acks + "]", extra=d)
-
-        consumerEntityIdList = system_test_utils.get_data_from_list_of_dicts( \
-                           clusterEntityConfigDictList, "role", "console_consumer", "entity_id")
-
-        matchingConsumerEntityId = None
-        for consumerEntityId in consumerEntityIdList:
-            consumerTopic = system_test_utils.get_data_by_lookup_keyval(testcaseEnv.testcaseConfigsList,
"entity_id", consumerEntityId, "topic")
-            if consumerTopic in topic:
-                matchingConsumerEntityId = consumerEntityId
-                break
-
-        if matchingConsumerEntityId is None:
-            break
-
-        producerLogPath     = get_testcase_config_log_dir_pathname(testcaseEnv, "producer_performance",
producerEntityId, "default")
-        producerLogPathName = producerLogPath + "/producer_performance.log"
-        producerMsgIdList   = get_message_id(producerLogPathName)
-        producerMsgIdSet    = set(producerMsgIdList)
-        logger.info("no. of unique messages on topic [" + topic + "] sent from publisher
 : " + str(len(producerMsgIdSet)), extra=d)
-        validationStatusDict["Unique messages from producer on [" + topic + "]"] = str(len(producerMsgIdSet))
-
-        consumerLogPath     = get_testcase_config_log_dir_pathname(testcaseEnv, "console_consumer",
matchingConsumerEntityId, "default")
-        for logFile in sorted(os.listdir(consumerLogPath)):
-            # only process log file: *.log
-            if logFile.endswith(".log"):
-                consumerLogPathName = consumerLogPath + "/" + logFile
-                consumerMsgIdList   = get_message_id(consumerLogPathName)
-                consumerMsgIdSet   = set(consumerMsgIdList)
-                missingMsgIdInConsumer = producerMsgIdSet - consumerMsgIdSet
-                msgIdMissingInConsumerLogPathName = get_testcase_config_log_dir_pathname(

-                    testcaseEnv, "console_consumer", matchingConsumerEntityId, "default")
+ \
-                    "/" + logFile + "_msg_id_missing_in_consumer.log"
-
-                outfile = open(msgIdMissingInConsumerLogPathName, "w")
-                for id in missingMsgIdInConsumer:
-                    outfile.write(id + "\n")
-                outfile.close()
-
-                logger.info("no. of unique messages on topic [" + topic + "] at " + logFile
+ " : " + str(len(consumerMsgIdSet)), extra=d)
-                validationStatusDict["Unique messages from consumer on [" + topic + "] at
" + logFile] = str(len(consumerMsgIdSet))
-
-                if acks == "-1" and len(missingMsgIdInConsumer) > 0:
-                    mismatchCount += 1
-                elif acks == "1" and len(missingMsgIdInConsumer) > 0:
-                    missingPercentage = len(missingMsgIdInConsumer) * 100 / len(producerMsgIdSet)
-                    logger.debug("missing percentage [" + str(missingPercentage) + "]", extra=d)
-                    if missingPercentage <= 1:
-                        logger.warn("Test case (acks == 1) passes with < 1% data loss
: [" + \
-                            str(len(missingMsgIdInConsumer)) + "] missing messages", extra=d)
-                    else:
-                        mismatchCount += 1
-
-        if mismatchCount == 0:
-            validationStatusDict["Validate for data matched on topic [" + topic + "]"] =
"PASSED"
-        else:
-            validationStatusDict["Validate for data matched on topic [" + topic + "]"] =
"FAILED"
-
 def get_controller_attributes(systemTestEnv, testcaseEnv):
 
     logger.info("Querying Zookeeper for Controller info ...", extra=d)
@@ -1917,7 +1863,7 @@ def get_controller_attributes(systemTestEnv, testcaseEnv):
                   "\"JAVA_HOME=" + javaHome,
                   kafkaRunClassBin + " org.apache.zookeeper.ZooKeeperMain",
                   "-server " + testcaseEnv.userDefinedEnvVarDict["sourceZkConnectStr"],
-                  "'get /controller' 2> /dev/null | tail -1\""]
+                  "get /controller 2> /dev/null | tail -1\""]
 
     cmdStr = " ".join(cmdStrList)
     logger.debug("executing command [" + cmdStr + "]", extra=d)
@@ -2007,6 +1953,8 @@ def getMinCommonStartingOffset(systemTestEnv, testcaseEnv, clusterName="source")
     return minCommonStartOffsetDict
 
 def validate_simple_consumer_data_matched_across_replicas(systemTestEnv, testcaseEnv):
+    logger.debug("#### Inside validate_simple_consumer_data_matched_across_replicas", extra=d)
+
     validationStatusDict        = testcaseEnv.validationStatusDict
     clusterEntityConfigDictList = systemTestEnv.clusterEntityConfigDictList
     consumerEntityIdList        = system_test_utils.get_data_from_list_of_dicts(
@@ -2014,101 +1962,100 @@ def validate_simple_consumer_data_matched_across_replicas(systemTestEnv,
testcas
     replicaFactor               = testcaseEnv.testcaseArgumentsDict["replica_factor"]
     numPartition                = testcaseEnv.testcaseArgumentsDict["num_partition"]
 
-    # Unique messages from producer on [test_1]  :  1500
-    # Unique messages from consumer on [test_1]  :  1500
+    for consumerEntityId in consumerEntityIdList:
 
-    # Unique messages from consumer on [test_1] at simple_consumer_test_1-0_r1.log  :  750
-    # Unique messages from consumer on [test_1] at simple_consumer_test_1-0_r2.log  :  750
-    # Unique messages from consumer on [test_1] at simple_consumer_test_1-0_r3.log  :  0
+        # get topic string from multi consumer "entity"
+        topicStr  = system_test_utils.get_data_by_lookup_keyval(testcaseEnv.testcaseConfigsList,
"entity_id", consumerEntityId, "topic")
 
-    # Unique messages from consumer on [test_1] at simple_consumer_test_1-1_r1.log  :  0
-    # Unique messages from consumer on [test_1] at simple_consumer_test_1-1_r2.log  :  750
-    # Unique messages from consumer on [test_1] at simple_consumer_test_1-1_r3.log  :  750
+        # the topic string could be multi topics separated by ','
+        topicList = topicStr.split(',')
 
-    # ==================================================
+        for topic in topicList:
+            logger.debug("working on topic : " + topic, extra=d)
+            consumerLogPath = get_testcase_config_log_dir_pathname(testcaseEnv, "console_consumer",
consumerEntityId, "default")
+
+            # keep track of total msg count across replicas for each topic-partition
+            # (should be greater than 0 for passing)
+            totalMsgCounter = 0
+
+            # keep track of the mismatch msg count for each topic-partition
+            # (should be equal to 0 for passing)
+            mismatchCounter = 0
+
+            replicaIdxMsgIdList = []
+            # replicaIdxMsgIdList :
+            # - This is a list of dictionaries of topic-partition (key)
+            #   mapping to list of MessageID in that topic-partition (val)
+            # - The list index is mapped to (replicaId - 1)
+            # [
+            #  // list index = 0 => replicaId = idx(0) + 1 = 1
+            #  {
+            #      "topic1-0" : [ "0000000001", "0000000002", "0000000003"],
+            #      "topic1-1" : [ "0000000004", "0000000005", "0000000006"]
+            #  },
+            #  // list index = 1 => replicaId = idx(1) + 1 = 2
+            #  {
+            #      "topic1-0" : [ "0000000001", "0000000002", "0000000003"],
+            #      "topic1-1" : [ "0000000004", "0000000005", "0000000006"]
+            #  }
+            # ]
+
+            # initialize replicaIdxMsgIdList
+            j = 0
+            while j < int(replicaFactor):
+                newDict = {}
+                replicaIdxMsgIdList.append(newDict)
+                j += 1
+
+            # retrieve MessageID from all simple consumer log4j files
+            for logFile in sorted(os.listdir(consumerLogPath)):
+
+                if logFile.startswith("simple_consumer_"+topic) and logFile.endswith(".log"):
+                    logger.debug("working on file : " + logFile, extra=d)
+                    matchObj    = re.match("simple_consumer_"+topic+"-(\d*)_r(\d*)\.log"
, logFile)
+                    partitionId = int(matchObj.group(1))
+                    replicaIdx  = int(matchObj.group(2))
+
+                    consumerLogPathName   = consumerLogPath + "/" + logFile
+                    consumerMsgIdList     = get_message_id(consumerLogPathName)
 
-    # Unique messages from producer on [test_2]  :  1000
-    # Unique messages from consumer on [test_2]  :  1000
+                    topicPartition = topic + "-" + str(partitionId)
+                    replicaIdxMsgIdList[replicaIdx - 1][topicPartition] = consumerMsgIdList
 
-    # Unique messages from consumer on [test_2] at simple_consumer_test_2-0_r1.log  :  500
-    # Unique messages from consumer on [test_2] at simple_consumer_test_2-0_r2.log  :  0
-    # Unique messages from consumer on [test_2] at simple_consumer_test_2-0_r3.log  :  500
+                    logger.info("no. of messages on topic [" + topic + "] at " + logFile
+ " : " + str(len(consumerMsgIdList)), extra=d)
+                    validationStatusDict["No. of messages from consumer on [" + topic + "]
at " + logFile] = str(len(consumerMsgIdList))
 
-    # Unique messages from consumer on [test_2] at simple_consumer_test_2-1_r1.log  :  500
-    # Unique messages from consumer on [test_2] at simple_consumer_test_2-1_r2.log  :  500
-    # Unique messages from consumer on [test_2] at simple_consumer_test_2-1_r3.log  :  0
+            # print replicaIdxMsgIdList
 
-    mismatchCounter = 0
-    for consumerEntityId in consumerEntityIdList:
+            # take the first dictionary of replicaIdxMsgIdList and compare with the rest
+            firstMsgIdDict = replicaIdxMsgIdList[0]
 
-        topic           = system_test_utils.get_data_by_lookup_keyval(testcaseEnv.testcaseConfigsList,
"entity_id", consumerEntityId, "topic")
-        consumerLogPath = get_testcase_config_log_dir_pathname(testcaseEnv, "console_consumer",
consumerEntityId, "default")
-
-        replicaIdxMsgCountDictList = []
-        # replicaIdxMsgCountDictList is being used as follows:
-        #
-        # the above replica message count will be organized as follows:
-        # index of the list would map to the partitionId
-        # each element in the list maps to the replicaIdx-MessageCount
-        # to validate that :
-        # 1. there should be "no. of broker" of non-zero message count and they are equal
-        # 2. there should be "no. of broker - replication factor" of zero count
-        # [{"1": "750", "2": "750", "3": "0"  },
-        #  {"1": "0"  , "2": "750", "3": "750"}]
-
-        j = 0
-        while j < int(numPartition):
-            newDict = {}
-            replicaIdxMsgCountDictList.append(newDict)
-            j += 1
-
-        for logFile in sorted(os.listdir(consumerLogPath)):
-
-            if logFile.startswith("simple_consumer_") and logFile.endswith(".log"):
-                matchObj    = re.match("simple_consumer_"+topic+"-(\d*)_r(\d*)\.log" , logFile)
-                partitionId = int(matchObj.group(1))
-                replicaIdx  = int(matchObj.group(2))
-
-                consumerLogPathName   = consumerLogPath + "/" + logFile
-                consumerMsgIdList     = get_message_id(consumerLogPathName)
-                consumerMsgIdSet      = set(consumerMsgIdList)
-
-                replicaIdxMsgCountDictList[partitionId][replicaIdx] = len(consumerMsgIdSet)
-
-                logger.info("no. of unique messages on topic [" + topic + "] at " + logFile
+ " : " + str(len(consumerMsgIdSet)), extra=d)
-                validationStatusDict["Unique messages from consumer on [" + topic + "] at
" + logFile] = str(len(consumerMsgIdSet))
-
-        pprint.pprint(replicaIdxMsgCountDictList)
-
-        partitionId = 0
-        while partitionId < int(numPartition):
-            zeroMsgCounter    = 0
-            nonZeroMsgCounter = 0
-            nonZeroMsgValue   = -1
-
-            for replicaIdx in sorted(replicaIdxMsgCountDictList[partitionId].iterkeys()):
-                if replicaIdxMsgCountDictList[partitionId][int(replicaIdx)] == 0:
-                    zeroMsgCounter += 1
-                else:
-                    if nonZeroMsgValue == -1:
-                        nonZeroMsgValue = replicaIdxMsgCountDictList[partitionId][int(replicaIdx)]
-                    else:
-                        if nonZeroMsgValue != replicaIdxMsgCountDictList[partitionId][int(replicaIdx)]:
-                            mismatchCounter += 1
-                    nonZeroMsgCounter += 1
-            partitionId += 1
+            # loop through all 'topic-partition' such as topic1-0, topic1-1, ...
+            for topicPartition in sorted(firstMsgIdDict.iterkeys()):
 
-            logger.info("topic " + topic + " : no. of brokers with zero msg count     : "
+ str(zeroMsgCounter), extra=d)
-            logger.info("topic " + topic + " : no. of brokers with non-zero msg count : "
+ str(nonZeroMsgCounter), extra=d)
-            logger.info("topic " + topic + " : non-zero brokers msg count             : "
+ str(nonZeroMsgValue), extra=d)
+                # compare all replicas' MessageID in corresponding topic-partition
+                for i in range(len(replicaIdxMsgIdList)):
+                    # skip the first dictionary
+                    if i == 0:
+                        totalMsgCounter += len(firstMsgIdDict[topicPartition])
+                        continue
 
-        if mismatchCounter == 0 and nonZeroMsgCounter > 0:
-            validationStatusDict["Validate for data matched on topic [" + topic + "] across
replicas"] = "PASSED"
-        else:
-            validationStatusDict["Validate for data matched on topic [" + topic + "] across
replicas"] = "FAILED"
+                    totalMsgCounter += len(replicaIdxMsgIdList[i][topicPartition])
+
+                    # get the count of mismatch MessageID between first MessageID list and
the other lists
+                    diffCount = system_test_utils.diff_lists(firstMsgIdDict[topicPartition],
replicaIdxMsgIdList[i][topicPartition])
+                    mismatchCounter += diffCount
+                    logger.info("Mismatch count of topic-partition [" + topicPartition +
"] in replica id [" + str(i+1) + "] : " + str(diffCount), extra=d)
+
+            if mismatchCounter == 0 and totalMsgCounter > 0:
+                validationStatusDict["Validate for data matched on topic [" + topic + "]
across replicas"] = "PASSED"
+            else:
+                validationStatusDict["Validate for data matched on topic [" + topic + "]
across replicas"] = "FAILED"
 
 
-def validate_data_matched_in_multi_topics_from_single_consumer_producer(systemTestEnv, testcaseEnv):
+def validate_data_matched_in_multi_topics_from_single_consumer_producer(systemTestEnv, testcaseEnv,
replicationUtils):
+    logger.debug("#### Inside validate_data_matched_in_multi_topics_from_single_consumer_producer",
extra=d)
+
     validationStatusDict        = testcaseEnv.validationStatusDict
     clusterEntityConfigDictList = systemTestEnv.clusterEntityConfigDictList
 
@@ -2140,6 +2087,7 @@ def validate_data_matched_in_multi_topics_from_single_consumer_producer(systemTe
 
         topicList = topicStr.split(',')
         for topic in topicList:
+            consumerDuplicateCount = 0
             msgIdMissingInConsumerLogPathName = get_testcase_config_log_dir_pathname( 
                                                 testcaseEnv, "console_consumer", matchingConsumerEntityId,
"default") \
                                                 + "/msg_id_missing_in_consumer_" + topic
+ ".log"
@@ -2148,10 +2096,11 @@ def validate_data_matched_in_multi_topics_from_single_consumer_producer(systemTe
             producerMsgIdSet   = set(producerMsgIdList)
             consumerMsgIdSet   = set(consumerMsgIdList)
 
-            missingMsgIdInConsumer = producerMsgIdSet - consumerMsgIdSet
+            consumerDuplicateCount = len(consumerMsgIdList) -len(consumerMsgIdSet)
+            missingUniqConsumerMsgId = system_test_utils.subtract_list(producerMsgIdSet,
consumerMsgIdSet)
 
             outfile = open(msgIdMissingInConsumerLogPathName, "w")
-            for id in missingMsgIdInConsumer:
+            for id in missingUniqConsumerMsgId:
                 outfile.write(id + "\n")
             outfile.close()
 
@@ -2160,14 +2109,21 @@ def validate_data_matched_in_multi_topics_from_single_consumer_producer(systemTe
             validationStatusDict["Unique messages from producer on [" + topic + "]"] = str(len(producerMsgIdSet))
             validationStatusDict["Unique messages from consumer on [" + topic + "]"] = str(len(consumerMsgIdSet))
 
-            if ( len(missingMsgIdInConsumer) == 0 and len(producerMsgIdSet) > 0 ):
+            missingPercentage = len(missingUniqConsumerMsgId) * 100.00 / len(producerMsgIdSet)
+            logger.info("Data loss threshold % : " + str(replicationUtils.ackOneDataLossThresholdPercent),
extra=d)
+            logger.warn("Data loss % on topic : " + topic + " : " + str(missingPercentage),
extra=d)
+
+            if ( len(missingUniqConsumerMsgId) == 0 and len(producerMsgIdSet) > 0 ):
                 validationStatusDict["Validate for data matched on topic [" + topic + "]"]
= "PASSED"
             elif (acks == "1"):
-                missingPercentage = len(missingMsgIdInConsumer) * 100 / len(producerMsgIdSet)
-                print "#### missing Percent : ", missingPercentage
-                if missingPercentage <= 1:
+                if missingPercentage <= replicationUtils.ackOneDataLossThresholdPercent:
                     validationStatusDict["Validate for data matched on topic [" + topic +
"]"] = "PASSED"
-                    logger.warn("Test case passes with less than 1% data loss : [" + str(len(missingMsgIdInConsumer))
+ "] missing messages", extra=d)
+                    logger.warn("Test case (Acks = 1) passes with less than " + str(replicationUtils.ackOneDataLossThresholdPercent)
\
+                        + "% data loss : [" + str(len(missingUniqConsumerMsgId)) + "] missing
messages", extra=d)
+                else:
+                    validationStatusDict["Validate for data matched on topic [" + topic +
"]"] = "FAILED"
+                    logger.error("Test case (Acks = 1) failed with more than " + str(replicationUtils.ackOneDataLossThresholdPercent)
\
+                        + "% data loss : [" + str(len(missingUniqConsumerMsgId)) + "] missing
messages", extra=d)
             else:
                 validationStatusDict["Validate for data matched on topic [" + topic + "]"]
= "FAILED"
                 logger.info("See " + msgIdMissingInConsumerLogPathName + " for missing MessageID",
extra=d)

http://git-wip-us.apache.org/repos/asf/kafka/blob/26c50fac/system_test/utils/replication_utils.py
----------------------------------------------------------------------
diff --git a/system_test/utils/replication_utils.py b/system_test/utils/replication_utils.py
index 3e8efad..cfd80b2 100644
--- a/system_test/utils/replication_utils.py
+++ b/system_test/utils/replication_utils.py
@@ -65,3 +65,6 @@ class ReplicationUtils(object):
         self.controllerAttributesDict["REGX_CONTROLLER_STARTUP_PATTERN"] = "\[(.*?)\] .*
\[Controller (.*?)\]: " + \
             self.controllerAttributesDict["CONTROLLER_STARTUP_COMPLETE_MSG"]
 
+        # Data Loss Percentage Threshold in Ack = 1 cases
+        self.ackOneDataLossThresholdPercent = 5.0
+

http://git-wip-us.apache.org/repos/asf/kafka/blob/26c50fac/system_test/utils/system_test_utils.py
----------------------------------------------------------------------
diff --git a/system_test/utils/system_test_utils.py b/system_test/utils/system_test_utils.py
index 65db5c5..50340f0 100644
--- a/system_test/utils/system_test_utils.py
+++ b/system_test/utils/system_test_utils.py
@@ -21,6 +21,7 @@
 # ===================================
 
 import copy
+import difflib
 import inspect
 import json
 import logging
@@ -554,5 +555,80 @@ def setup_remote_hosts_with_testsuite_level_cluster_config(systemTestEnv,
testMo
         sys.exit(1)
     print
 
-
+# =================================================
+# lists_diff_count
+# - find the no. of different items in both lists
+# - both lists need not be sorted
+# - input lists won't be changed
+# =================================================
+def lists_diff_count(a, b):
+    c = list(b)
+    d = []
+    for item in a:
+        try:
+            c.remove(item)
+        except:
+            d.append(item)
+
+    if len(d) > 0:
+        print "#### Mismatch MessageID"
+        print d
+
+    return len(c) + len(d)
+
+# =================================================
+# subtract_list
+# - subtract items in listToSubtract from mainList
+#   and return the resulting list
+# - both lists need not be sorted
+# - input lists won't be changed
+# =================================================
+def subtract_list(mainList, listToSubtract):
+    remainingList = list(mainList)
+    for item in listToSubtract:
+        try:
+            remainingList.remove(item)
+        except:
+            pass
+    return remainingList
+
+# =================================================
+# diff_lists
+# - find the diff of 2 lists and return the 
+#   total no. of mismatch from both lists
+# - diff of both lists includes:
+#   - no. of items mismatch
+#   - ordering of the items
+#
+# sample lists:
+# a = ['8','4','3','2','1']
+# b = ['8','3','4','2','1']
+#
+# difflib will return the following:
+#   8
+# + 3
+#   4
+# - 3
+#   2
+#   1
+#
+# diff_lists(a,b) returns 2 and prints the following:
+# #### only in seq 2 :  + 3
+# #### only in seq 1 :  - 3
+# =================================================
+def diff_lists(a, b):
+    mismatchCount = 0
+    d = difflib.Differ()
+    diff = d.compare(a,b)
+
+    for item in diff:
+        result = item[0:1].strip()
+        if len(result) > 0:
+            mismatchCount += 1
+            if '-' in result:
+                logger.debug("#### only in seq 1 : " + item, extra=d)
+            elif '+' in result:
+                logger.debug("#### only in seq 2 : " + item, extra=d)
+
+    return mismatchCount
 


Mime
View raw message