[ 
https://issues.apache.org/jira/browse/FLINK-8306?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16319763#comment-16319763
 ] 

ASF GitHub Bot commented on FLINK-8306:
---------------------------------------

Github user tzulitai commented on a diff in the pull request:

    https://github.com/apache/flink/pull/5200#discussion_r160598939
  
    --- Diff: 
flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java
 ---
    @@ -89,30 +90,44 @@
        @SuppressWarnings("unchecked")
        public void testEitherWatermarkExtractor() {
                try {
    -                   new 
DummyFlinkKafkaConsumer<String>(mock(AbstractFetcher.class), 
mock(AbstractPartitionDiscoverer.class), false)
    +                   new DummyFlinkKafkaConsumer<String>(
    +                                   mock(AbstractFetcher.class),
    +                                   mock(AbstractPartitionDiscoverer.class),
    +                                   mock(KafkaOffsetCommitter.class),
    +                                   false)
                                
.assignTimestampsAndWatermarks((AssignerWithPeriodicWatermarks<String>) null);
                        fail();
                } catch (NullPointerException ignored) {}
     
                try {
    -                   new 
DummyFlinkKafkaConsumer<String>(mock(AbstractFetcher.class), 
mock(AbstractPartitionDiscoverer.class), false)
    +                   new DummyFlinkKafkaConsumer<String>(
    +                                   mock(AbstractFetcher.class),
    +                                   mock(AbstractPartitionDiscoverer.class),
    +                                   mock(KafkaOffsetCommitter.class),
    +                                   false)
                                
.assignTimestampsAndWatermarks((AssignerWithPunctuatedWatermarks<String>) null);
                        fail();
                } catch (NullPointerException ignored) {}
     
                final AssignerWithPeriodicWatermarks<String> periodicAssigner = 
mock(AssignerWithPeriodicWatermarks.class);
                final AssignerWithPunctuatedWatermarks<String> 
punctuatedAssigner = mock(AssignerWithPunctuatedWatermarks.class);
     
    -           DummyFlinkKafkaConsumer<String> c1 =
    -                   new 
DummyFlinkKafkaConsumer<>(mock(AbstractFetcher.class), 
mock(AbstractPartitionDiscoverer.class), false);
    +           DummyFlinkKafkaConsumer<String> c1 = new 
DummyFlinkKafkaConsumer<>(
    +                           mock(AbstractFetcher.class),
    +                           mock(AbstractPartitionDiscoverer.class),
    +                           mock(KafkaOffsetCommitter.class),
    --- End diff --
    
    Will do.


> FlinkKafkaConsumerBaseTest has invalid mocks on final methods
> -------------------------------------------------------------
>
>                 Key: FLINK-8306
>                 URL: https://issues.apache.org/jira/browse/FLINK-8306
>             Project: Flink
>          Issue Type: Bug
>          Components: Kafka Connector, Tests
>            Reporter: Tzu-Li (Gordon) Tai
>            Assignee: Tzu-Li (Gordon) Tai
>            Priority: Critical
>             Fix For: 1.5.0, 1.4.1
>
>
> The {{FlinkKafkaConsumerBaseTest}} has invalid mocks on a final 
> {{AbstractFetcher::commitInternalOffsetsToKafka(...)}} method. While an easy 
> fix would be to simply make that method non-final, that is not ideal since it 
> would be best that the method is left final to prevent overrides in 
> subclasses.
> This suggests that offset committing functionality is too tightly coupled 
> with the {{AbstractFetcher}}, making it hard to perform concise tests to 
> verify offset committing.
> I suggest that we decouple record fetching and offset committing as separate 
> services behind different interfaces. We should introduce a new interface, 
> say {{KafkaOffsetCommitter}}, and test against that instead. Initially, we 
> can simply let {{AbstractFetcher}} implement {{KafkaOffsetCommitter}}.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

Reply via email to