From dev-return-61028-apmail-storm-dev-archive=storm.apache.org@storm.apache.org Sun Nov 10 14:30:29 2019 Return-Path: X-Original-To: apmail-storm-dev-archive@minotaur.apache.org Delivered-To: apmail-storm-dev-archive@minotaur.apache.org Received: from mail.apache.org (hermes.apache.org [207.244.88.153]) by minotaur.apache.org (Postfix) with SMTP id A361E105BB for ; Sun, 10 Nov 2019 14:30:29 +0000 (UTC) Received: (qmail 37759 invoked by uid 500); 10 Nov 2019 14:30:27 -0000 Delivered-To: apmail-storm-dev-archive@storm.apache.org Received: (qmail 37708 invoked by uid 500); 10 Nov 2019 14:30:27 -0000 Mailing-List: contact dev-help@storm.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: dev@storm.apache.org Delivered-To: mailing list dev@storm.apache.org Received: (qmail 37696 invoked by uid 99); 10 Nov 2019 14:30:27 -0000 Received: from ec2-52-202-80-70.compute-1.amazonaws.com (HELO gitbox.apache.org) (52.202.80.70) by apache.org (qpsmtpd/0.29) with ESMTP; Sun, 10 Nov 2019 14:30:27 +0000 From: GitBox To: dev@storm.apache.org Subject: [GitHub] [storm] OliverMD commented on a change in pull request #3164: STORM-3529: Catch and Log Kafka RetriableException Message-ID: <157339622763.10632.2654862570128209090.gitbox@gitbox.apache.org> Date: Sun, 10 Nov 2019 14:30:27 -0000 Content-Type: text/plain; charset=utf-8 Content-Transfer-Encoding: 8bit OliverMD commented on a change in pull request #3164: STORM-3529: Catch and Log Kafka RetriableException URL: https://github.com/apache/storm/pull/3164#discussion_r344495310 ########## File path: external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/KafkaSpoutSingleTopicTest.java ########## @@ -428,4 +427,33 @@ public void testOffsetMetrics() throws Exception { assertEquals(offsetMetric.get(SingleTopicKafkaSpoutConfiguration.TOPIC+"/totalLatestCompletedOffset").longValue(), 10); assertEquals(offsetMetric.get(SingleTopicKafkaSpoutConfiguration.TOPIC+"/totalSpoutLag").longValue(), 0); } + + @Test + public void testOffsetMetricsReturnsNullWhenRetriableExceptionThrown() throws Exception { + final int messageCount = 10; + prepareSpout(messageCount); + + Map offsetMetric = (Map) spout.getKafkaOffsetMetric().getValueAndReset(); + assertEquals(offsetMetric.get(SingleTopicKafkaSpoutConfiguration.TOPIC+"/totalEarliestTimeOffset").longValue(), 0); + // the offset of the last available message + 1. + assertEquals(offsetMetric.get(SingleTopicKafkaSpoutConfiguration.TOPIC+"/totalLatestTimeOffset").longValue(), 10); + assertEquals(offsetMetric.get(SingleTopicKafkaSpoutConfiguration.TOPIC+"/totalRecordsInPartitions").longValue(), 10); + assertEquals(offsetMetric.get(SingleTopicKafkaSpoutConfiguration.TOPIC+"/totalLatestEmittedOffset").longValue(), 0); + assertEquals(offsetMetric.get(SingleTopicKafkaSpoutConfiguration.TOPIC+"/totalLatestCompletedOffset").longValue(), 0); + //totalSpoutLag = totalLatestTimeOffset-totalLatestCompletedOffset + assertEquals(offsetMetric.get(SingleTopicKafkaSpoutConfiguration.TOPIC+"/totalSpoutLag").longValue(), 10); + + //Emit all messages and check that they are emitted. Ack the messages too + for (int i = 0; i < messageCount; i++) { + nextTuple_verifyEmitted_ack_resetCollector(i); + } + + commitAndVerifyAllMessagesCommitted(messageCount); Review comment: @srdo Do you think it would make sense to not include this as the test above already tests this? ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: users@infra.apache.org With regards, Apache Git Services