[GitHub] flink pull request #2430: [FLINK-3874] Rewrite Kafka JSON Table sink tests

[fhueske <gi...@git.apache.org>]

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

    https://github.com/apache/flink/pull/2430#discussion_r79390399
  
    --- Diff: flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSinkTestBase.java ---
    @@ -17,123 +17,60 @@
      */
     package org.apache.flink.streaming.connectors.kafka;
     
    -import org.apache.flink.api.common.functions.RichMapFunction;
    -import org.apache.flink.api.common.restartstrategy.RestartStrategies;
     import org.apache.flink.api.common.typeinfo.TypeInformation;
     import org.apache.flink.api.table.Row;
    +import org.apache.flink.api.table.typeutils.RowTypeInfo;
     import org.apache.flink.streaming.api.datastream.DataStream;
    -import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    -import org.apache.flink.streaming.api.functions.sink.SinkFunction;
    -import org.apache.flink.streaming.api.functions.source.SourceFunction;
     import org.apache.flink.streaming.connectors.kafka.internals.TypeUtil;
     import org.apache.flink.streaming.connectors.kafka.partitioner.KafkaPartitioner;
    -import org.apache.flink.streaming.util.serialization.DeserializationSchema;
    -import org.apache.flink.test.util.SuccessException;
     import org.junit.Test;
     
     import java.io.Serializable;
    -import java.util.HashSet;
     import java.util.Properties;
     
    -import static org.apache.flink.test.util.TestUtils.tryExecute;
    +import static org.junit.Assert.*;
    +import static org.mockito.Mockito.mock;
    +import static org.mockito.Mockito.verify;
     
    -public abstract class KafkaTableSinkTestBase extends KafkaTestBase implements Serializable {
    +public abstract class KafkaTableSinkTestBase implements Serializable {
     
    -	protected final static String TOPIC = "customPartitioningTestTopic";
    -	protected final static int PARALLELISM = 1;
    +	protected final static String TOPIC = "testTopic";
     	protected final static String[] FIELD_NAMES = new String[] {"field1", "field2"};
     	protected final static TypeInformation[] FIELD_TYPES = TypeUtil.toTypeInfo(new Class[] {Integer.class, String.class});
     
    +	protected FlinkKafkaProducerBase<Row> kafkaProducer = mock(FlinkKafkaProducerBase.class);
    +
     	@Test
     	public void testKafkaTableSink() throws Exception {
    -		LOG.info("Starting KafkaTableSinkTestBase.testKafkaTableSink()");
    -
    -		createTestTopic(TOPIC, PARALLELISM, 1);
    -		StreamExecutionEnvironment env = createEnvironment();
    -
    -		createProducingTopology(env);
    -		createConsumingTopology(env);
    -
    -		tryExecute(env, "custom partitioning test");
    -		deleteTestTopic(TOPIC);
    -		LOG.info("Finished KafkaTableSinkTestBase.testKafkaTableSink()");
    -	}
    -
    -	private StreamExecutionEnvironment createEnvironment() {
    -		StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment("localhost", flinkPort);
    -		env.setRestartStrategy(RestartStrategies.noRestart());
    -		env.getConfig().disableSysoutLogging();
    -		return env;
    -	}
    -
    -	private void createProducingTopology(StreamExecutionEnvironment env) {
    -		DataStream<Row> stream = env.addSource(new SourceFunction<Row>() {
    -			private boolean running = true;
    -
    -			@Override
    -			public void run(SourceContext<Row> ctx) throws Exception {
    -				long cnt = 0;
    -				while (running) {
    -					Row row = new Row(2);
    -					row.setField(0, cnt);
    -					row.setField(1, "kafka-" + cnt);
    -					ctx.collect(row);
    -					cnt++;
    -				}
    -			}
    +		DataStream dataStream = mock(DataStream.class);
    +		KafkaTableSink kafkaTableSink = createTableSink();
    +		kafkaTableSink.emitDataStream(dataStream);
     
    -			@Override
    -			public void cancel() {
    -				running = false;
    -			}
    -		})
    -		.setParallelism(1);
    -
    -		KafkaTableSink kafkaTableSinkBase = createTableSink();
    -
    -		kafkaTableSinkBase.emitDataStream(stream);
    +		verify(dataStream).addSink(kafkaProducer);
    --- End diff --
    
    This test checks that `addSink` was called with a `FlinkKafkaProducerBase` object. I think it would be good to extend the test to make sure that
    - the right producer was used (Kafka 0.8, 0.9)
    - the correct topic is fetched
    - the properties are correctly passed on
    - the correct serialization scheme is used
    - the correct partitioner is used
    
    Would that be possible?


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---