diff --git a/system_test/mirror_maker_testsuite/mirror_maker_test.py b/system_test/mirror_maker_testsuite/mirror_maker_test.py index 48b0d25..70db79b 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, ... # ============================================================= @@ -283,7 +285,7 @@ class MirrorMakerTest(ReplicationUtils, SetupUtils): # ============================================= 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_simple_consumer_data_matched(self.systemTestEnv, self.testcaseEnv, replicationUtils) kafka_system_test_utils.validate_broker_log_segment_checksum(self.systemTestEnv, self.testcaseEnv) kafka_system_test_utils.validate_broker_log_segment_checksum(self.systemTestEnv, self.testcaseEnv, "target") diff --git a/system_test/replication_testsuite/replica_basic_test.py b/system_test/replication_testsuite/replica_basic_test.py index 3fc47d9..a89e96a 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, ... # ============================================================= @@ -424,15 +426,15 @@ class ReplicaBasicTest(ReplicationUtils, SetupUtils): 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 # ============================================= diff --git a/system_test/utils/kafka_system_test_utils.py b/system_test/utils/kafka_system_test_utils.py index 9411405..2bbfc23 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,7 +1234,8 @@ def validate_data_matched(systemTestEnv, testcaseEnv): producerMsgIdSet = set(producerMsgIdList) consumerMsgIdSet = set(consumerMsgIdList) - missingMsgIdInConsumer = producerMsgIdSet - consumerMsgIdSet + consumerDuplicateCount = len(consumerMsgIdList) -len(consumerMsgIdSet) + missingMsgIdInConsumer = system_test_utils.subtract_list(producerMsgIdSet, consumerMsgIdList) outfile = open(msgIdMissingInConsumerLogPathName, "w") for id in missingMsgIdInConsumer: @@ -1237,24 +1243,32 @@ def validate_data_matched(systemTestEnv, testcaseEnv): outfile.close() logger.info("no. of unique messages on topic [" + topic + "] sent from publisher : " + str(len(producerMsgIdSet)), extra=d) - logger.info("no. of unique messages on topic [" + topic + "] received by consumer : " + str(len(consumerMsgIdSet)), extra=d) + logger.info("no. of messages on topic [" + topic + "] received by consumer : " + str(len(consumerMsgIdList)), extra=d) validationStatusDict["Unique messages from producer on [" + topic + "]"] = str(len(producerMsgIdSet)) - validationStatusDict["Unique messages from consumer on [" + topic + "]"] = str(len(consumerMsgIdSet)) + validationStatusDict["No. of messages from consumer on [" + topic + "]"] = str(len(consumerMsgIdList)) - if ( len(missingMsgIdInConsumer) == 0 and len(producerMsgIdSet) > 0 ): + missingPercentage = len(missingMsgIdInConsumer) * 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(missingMsgIdInConsumer) == 0 and len(producerMsgIdSet) > 0 and consumerDuplicateCount == 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(missingMsgIdInConsumer)) + "] 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(missingMsgIdInConsumer)) + "] 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,7 +1840,9 @@ def start_simple_consumer(systemTestEnv, testcaseEnv, minStartingOffsetDict=None partitionId += 1 replicaIndex += 1 -def validate_simple_consumer_data_matched(systemTestEnv, testcaseEnv): +def validate_simple_consumer_data_matched(systemTestEnv, testcaseEnv, replicationUtils): + logger.debug("#### Inside validate_simple_consumer_data_matched", extra=d) + validationStatusDict = testcaseEnv.validationStatusDict clusterEntityConfigDictList = systemTestEnv.clusterEntityConfigDictList @@ -1831,6 +1850,7 @@ def validate_simple_consumer_data_matched(systemTestEnv, testcaseEnv): consumerCfgList = system_test_utils.get_dict_from_list_of_dicts(clusterEntityConfigDictList, "role", "console_consumer") mismatchCount = 0 + consumerDuplicateCount = 0 for prodPerfCfg in prodPerfCfgList: producerEntityId = prodPerfCfg["entity_id"] @@ -1864,8 +1884,12 @@ def validate_simple_consumer_data_matched(systemTestEnv, testcaseEnv): if logFile.endswith(".log"): consumerLogPathName = consumerLogPath + "/" + logFile consumerMsgIdList = get_message_id(consumerLogPathName) - consumerMsgIdSet = set(consumerMsgIdList) - missingMsgIdInConsumer = producerMsgIdSet - consumerMsgIdSet + consumerMsgIdSet = set(consumerMsgIdList) + consumerDuplicateCount = len(consumerMsgIdList) - len(consumerMsgIdSet) + + # producer MessageID are deduped but consumer's are not + # subtract all consumer MessageID from producer's and see if any items are left + missingMsgIdInConsumer = system_test_utils.subtract_list(producerMsgIdSet, consumerMsgIdList) msgIdMissingInConsumerLogPathName = get_testcase_config_log_dir_pathname( testcaseEnv, "console_consumer", matchingConsumerEntityId, "default") + \ "/" + logFile + "_msg_id_missing_in_consumer.log" @@ -1875,24 +1899,26 @@ def validate_simple_consumer_data_matched(systemTestEnv, testcaseEnv): 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)) + 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)) + + missingPercentage = len(missingMsgIdInConsumer) * 100.00 / len(producerMsgIdSet) + logger.info("Data loss threshold % : " + str(replicationUtils.ackOneDataLossThresholdPercent), extra=d) + logger.info("Data loss % on topic " + topic + " : " + str(missingPercentage), extra=d) 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) + if missingPercentage <= replicationUtils.ackOneDataLossThresholdPercent: + logger.warn("Test case (Acks = 1) passes with less than " + str(replicationUtils.ackOneDataLossThresholdPercent) \ + + "% 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" + if mismatchCount == 0 and len(producerMsgIdSet) > 0 and consumerDuplicateCount == 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): @@ -1917,7 +1943,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 +2033,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 +2042,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 - - # 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 - - # 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 - - # ================================================== - - # Unique messages from producer on [test_2] : 1000 - # Unique messages from consumer on [test_2] : 1000 - - # 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 + for consumerEntityId in consumerEntityIdList: - # 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 + # get topic string from multi consumer "entity" + topicStr = system_test_utils.get_data_by_lookup_keyval(testcaseEnv.testcaseConfigsList, "entity_id", consumerEntityId, "topic") - mismatchCounter = 0 - for consumerEntityId in consumerEntityIdList: + # the topic string could be multi topics separated by ',' + topicList = topicStr.split(',') - 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 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) - for logFile in sorted(os.listdir(consumerLogPath)): + topicPartition = topic + "-" + str(partitionId) + replicaIdxMsgIdList[replicaIdx - 1][topicPartition] = consumerMsgIdList - 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)) + 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)) - consumerLogPathName = consumerLogPath + "/" + logFile - consumerMsgIdList = get_message_id(consumerLogPathName) - consumerMsgIdSet = set(consumerMsgIdList) + # print replicaIdxMsgIdList - replicaIdxMsgCountDictList[partitionId][replicaIdx] = len(consumerMsgIdSet) + # take the first dictionary of replicaIdxMsgIdList and compare with the rest + firstMsgIdDict = replicaIdxMsgIdList[0] - 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)) + # loop through all 'topic-partition' such as topic1-0, topic1-1, ... + for topicPartition in sorted(firstMsgIdDict.iterkeys()): - pprint.pprint(replicaIdxMsgCountDictList) + # 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 - partitionId = 0 - while partitionId < int(numPartition): - zeroMsgCounter = 0 - nonZeroMsgCounter = 0 - nonZeroMsgValue = -1 + totalMsgCounter += len(replicaIdxMsgIdList[i][topicPartition]) - 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 + # get the count of mismatch MessageID between first MessageID list and the other lists + diffCount = system_test_utils.lists_diff_count(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) - 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) + 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" - 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" +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) -def validate_data_matched_in_multi_topics_from_single_consumer_producer(systemTestEnv, testcaseEnv): validationStatusDict = testcaseEnv.validationStatusDict clusterEntityConfigDictList = systemTestEnv.clusterEntityConfigDictList @@ -2140,6 +2167,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,7 +2176,8 @@ 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) + missingMsgIdInConsumer = system_test_utils.subtract_list(producerMsgIdSet, consumerMsgIdList) outfile = open(msgIdMissingInConsumerLogPathName, "w") for id in missingMsgIdInConsumer: @@ -2156,18 +2185,21 @@ def validate_data_matched_in_multi_topics_from_single_consumer_producer(systemTe outfile.close() logger.info("no. of unique messages on topic [" + topic + "] sent from publisher : " + str(len(producerMsgIdSet)), extra=d) - logger.info("no. of unique messages on topic [" + topic + "] received by consumer : " + str(len(consumerMsgIdSet)), extra=d) + logger.info("no. of messages on topic [" + topic + "] received by consumer : " + str(len(consumerMsgIdList)), extra=d) validationStatusDict["Unique messages from producer on [" + topic + "]"] = str(len(producerMsgIdSet)) - validationStatusDict["Unique messages from consumer on [" + topic + "]"] = str(len(consumerMsgIdSet)) + validationStatusDict["No. of messages from consumer on [" + topic + "]"] = str(len(consumerMsgIdList)) + + missingPercentage = len(missingMsgIdInConsumer) * 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(missingMsgIdInConsumer) == 0 and len(producerMsgIdSet) > 0 ): + if ( len(missingMsgIdInConsumer) == 0 and len(producerMsgIdSet) > 0 and consumerDuplicateCount == 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(missingMsgIdInConsumer)) + "] missing messages", extra=d) else: validationStatusDict["Validate for data matched on topic [" + topic + "]"] = "FAILED" logger.info("See " + msgIdMissingInConsumerLogPathName + " for missing MessageID", extra=d) 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 + diff --git a/system_test/utils/system_test_utils.py b/system_test/utils/system_test_utils.py index 65db5c5..33b1bf0 100644 --- a/system_test/utils/system_test_utils.py +++ b/system_test/utils/system_test_utils.py @@ -554,5 +554,39 @@ 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