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..4316fc3 100644 --- a/system_test/utils/kafka_system_test_utils.py +++ b/system_test/utils/kafka_system_test_utils.py @@ -1189,7 +1189,9 @@ 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 @@ -1198,8 +1200,9 @@ def validate_data_matched(systemTestEnv, testcaseEnv): 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") + #topic = testcaseEnv.producerTopicsString + 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 +1210,15 @@ 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") + #consumerTopic = testcaseEnv.consumerTopicsString 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( \ @@ -1241,20 +1246,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)) + missingPercentage = len(missingMsgIdInConsumer) * 100.00 / len(producerMsgIdSet) + logger.info("Data Loss Threshold % : " + str(replicationUtils.ackOneDataLossThresholdPercent), extra=d) + logger.warn("Data Loss Current % : " + str(missingPercentage), extra=d) + if ( len(missingMsgIdInConsumer) == 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(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 +1558,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 +1629,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 +1839,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 @@ -1878,21 +1896,23 @@ def validate_simple_consumer_data_matched(systemTestEnv, testcaseEnv): 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)) + missingPercentage = len(missingMsgIdInConsumer) * 100.00 / len(producerMsgIdSet) + logger.info("Data Loss Threshold % : " + str(replicationUtils.ackOneDataLossThresholdPercent), extra=d) + logger.warn("Data Loss Current % : " + 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: + 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 +1937,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 +2027,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,6 +2036,8 @@ def validate_simple_consumer_data_matched_across_replicas(systemTestEnv, testcas replicaFactor = testcaseEnv.testcaseArgumentsDict["replica_factor"] numPartition = testcaseEnv.testcaseArgumentsDict["num_partition"] + # Validation output may look like this: + # ===================================== # Unique messages from producer on [test_1] : 1500 # Unique messages from consumer on [test_1] : 1500 @@ -2025,90 +2049,94 @@ def validate_simple_consumer_data_matched_across_replicas(systemTestEnv, testcas # 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 - - # 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 - mismatchCounter = 0 for consumerEntityId in consumerEntityIdList: - 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 + # get topic string from multi consumer "entity" + topicStr = system_test_utils.get_data_by_lookup_keyval(testcaseEnv.testcaseConfigsList, "entity_id", consumerEntityId, "topic") + # the topic string could be multi topics separated by ',' + topicList = topicStr.split(',') + for topic in topicList: - for logFile in sorted(os.listdir(consumerLogPath)): + logger.debug("working on topic : " + topic, extra=d) + consumerLogPath = get_testcase_config_log_dir_pathname(testcaseEnv, "console_consumer", consumerEntityId, "default") - 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)) + 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"}] - consumerLogPathName = consumerLogPath + "/" + logFile - consumerMsgIdList = get_message_id(consumerLogPathName) - consumerMsgIdSet = set(consumerMsgIdList) + j = 0 + while j < int(numPartition): + newDict = {} + replicaIdxMsgCountDictList.append(newDict) + j += 1 - replicaIdxMsgCountDictList[partitionId][replicaIdx] = len(consumerMsgIdSet) + for logFile in sorted(os.listdir(consumerLogPath)): - 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 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)) - pprint.pprint(replicaIdxMsgCountDictList) + consumerLogPathName = consumerLogPath + "/" + logFile + consumerMsgIdList = get_message_id(consumerLogPathName) - partitionId = 0 - while partitionId < int(numPartition): - zeroMsgCounter = 0 - nonZeroMsgCounter = 0 - nonZeroMsgValue = -1 + # get the no. of messages received in each replica (no need to dedupe) + replicaIdxMsgCountDictList[partitionId][replicaIdx] = len(consumerMsgIdList) - for replicaIdx in sorted(replicaIdxMsgCountDictList[partitionId].iterkeys()): - if replicaIdxMsgCountDictList[partitionId][int(replicaIdx)] == 0: - zeroMsgCounter += 1 - else: - if nonZeroMsgValue == -1: - nonZeroMsgValue = replicaIdxMsgCountDictList[partitionId][int(replicaIdx)] + logger.info("no. of unique messages on topic [" + topic + "] at " + logFile + " : " + str(len(consumerMsgIdList)), extra=d) + validationStatusDict["Unique messages from consumer on [" + topic + "] at " + logFile] = str(len(consumerMsgIdList)) + + pprint.pprint(replicaIdxMsgCountDictList) + + partitionId = 0 + while partitionId < int(numPartition): + # numPartition is declared in testcase_xxxx_properties.json + # replicaIdxMsgCountDictList may look like this: + # [{1: 700, 2: 700, 3: 700}, <- partitionId = 0 + # {1: 700, 2: 700, 3: 700}, <- partitionId = 1 + # {1: 600, 2: 600, 3: 600}] <- partitionId = 2 + zeroMsgCounter = 0 + nonZeroMsgCounter = 0 + nonZeroMsgValue = -1 + + for replicaIdx in sorted(replicaIdxMsgCountDictList[partitionId].iterkeys()): + # going through each dictionary which looks like this: {1: 700, 2: 700, 3: 700} + if replicaIdxMsgCountDictList[partitionId][int(replicaIdx)] == 0: + zeroMsgCounter += 1 else: - if nonZeroMsgValue != replicaIdxMsgCountDictList[partitionId][int(replicaIdx)]: - mismatchCounter += 1 - nonZeroMsgCounter += 1 - partitionId += 1 + if nonZeroMsgValue == -1: + # save the no. of messages received from the first replica + nonZeroMsgValue = replicaIdxMsgCountDictList[partitionId][int(replicaIdx)] + else: + if nonZeroMsgValue != replicaIdxMsgCountDictList[partitionId][int(replicaIdx)]: + # found mismatch no. of messages across replicas for a certain topic-partition + mismatchCounter += 1 + nonZeroMsgCounter += 1 + partitionId += 1 - 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) + 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 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" + 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): +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 @@ -2160,14 +2188,17 @@ 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)) + missingPercentage = len(missingMsgIdInConsumer) * 100.00 / len(producerMsgIdSet) + logger.info("Data Loss Threshold % : " + str(replicationUtils.ackOneDataLossThresholdPercent), extra=d) + logger.warn("Data Loss Current % : " + str(missingPercentage), extra=d) + if ( len(missingMsgIdInConsumer) == 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(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 +