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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
