KAFKA-5520: KIP-171 - Extend Consumer Group Reset Offset for Stream Application - Updated by jeqo · Pull Request #4159 · apache/kafka

jeqo · 2017-10-30T13:24:29Z

KIP: https://cwiki.apache.org/confluence/display/KAFKA/KIP-171+-+Extend+Consumer+Group+Reset+Offset+for+Stream+Application

Merge changes from KIP-198

Ref: #3831

mjsax

Some initial comments. Did not yet have a look at the test (it seems, there is a lot of redundancy in the test though -- could we refactor to share more code?)

mjsax · 2017-11-10T23:21:47Z

+        }
+    }
+
+    public Long getDateTime(String ts) throws ParseException {


Can't this be package private? (add comment // visible for testing)

nit: ts -> timestamp

Yes, but test is in another package/module. Maybe once it is moved to streams we can improve visibility of this method. I will add the comment.

mjsax · 2017-11-10T23:23:31Z

+    }
+
+    public Long getDateTime(String ts) throws ParseException {
+        if (!(ts.split("T")[1].contains("+") || ts.split("T")[1].contains("-") || ts.split("T")[1].contains("Z"))) {


This could throw a NPE. I think we should guard against this, and throw ParseException if NPE happens

ts.split("T") is redundant and should be extracted into a variable.

Agree. I will reformat this.

mjsax · 2017-11-10T23:23:44Z

+        Date date;
+        try {
+            date = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSSXXX").parse(ts);
+        } catch (ParseException e) {


nit: add final

Agree. I'll reformat.

mjsax · 2017-11-10T23:27:42Z

+    }
+
+    @Test
+    public void testDateTimeFormats() throws ParseException {


this should be multiple tests (one test method per scenario). Also, use better method names, eg, shouldAcceptValidDateFormats, shouldThrowOnInvalidDateFormat etc

Yes, I will add them.

mjsax · 2017-11-10T23:29:07Z

+        try {
+            invokeGetDateTimeMethod(new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss"));
+            fail("Call to getDateTime should fail");
+        } catch (Exception e) {


nit: add final (same below)

Not sure where the final should be. Maybe you're referring to the next comment?

} catch (final Exception e) {

mjsax · 2017-11-10T23:29:15Z

+        }
+    }
+
+    private void invokeGetDateTimeMethod(SimpleDateFormat format) throws ParseException {


mjsax · 2017-11-10T23:34:20Z

+        final Map<TopicPartition, Long> topicPartitionAndOffset = new HashMap<>();
+        for (final String line : resetPlanCsv.split("\n")) {
+            final String[] parts = line.split(",");
+            final String topic = parts[0];


Should we guard against NPE? (same blow)

Yeap, I'll refactor.

mjsax · 2017-11-10T23:34:56Z

+        return date.getTime();
+    }
+
+    private Map<TopicPartition, Long> parseResetPlan(String resetPlanCsv) {


nit: add final

mjsax · 2017-11-10T23:36:31Z

+                                                       Map<TopicPartition, Long> beginningOffsets,
+                                                       Map<TopicPartition, Long> endOffsets) {
+        Map<TopicPartition, Long> validatedTopicPartitionsOffsets = new HashMap<>();
+        for (final TopicPartition topicPartition : inputTopicPartitionsAndOffset.keySet()) {


nit: As key and value is accessed, we might want to iterate throw the .entrySet instead of keySet

mjsax · 2017-11-10T23:37:13Z

+        return topicPartitionAndOffset;
+    }
+
+    private Map<TopicPartition, Long> checkOffsetRange(Map<TopicPartition, Long> inputTopicPartitionsAndOffset,


final (3x)

asfgit · 2017-11-14T14:44:36Z