Question: Right now my flink code is processing a file and sinking the data on kafka topic with 1 partition.

Now I have a topic with 2 partition and I want flink code to sink data on those 2 partition using DefaultPartitioner.

Could you help me with that.

Here is the code snippet of my current code:

```
DataStream<String> speStream = inputStream..map(new MapFunction<Row, String>(){....}
Properties props = Producer.getProducerConfig(propertiesFilePath);
speStream.addSink(new FlinkKafkaProducer011(kafkaTopicName, new KeyedSerializationSchemaWrapper<>(new SimpleStringSchema()), props, FlinkKafkaProducer011.Semantic.EXACTLY_ONCE));

```

Answer: 
Solved this by changing the flinkproducer to 

```
 speStream.addSink(new FlinkKafkaProducer011(kafkaTopicName,new SimpleStringSchema(), 
 props));

```

earlier i was using 

```
speStream.addSink(new FlinkKafkaProducer011(kafkaTopicName,
new KeyedSerializationSchemaWrapper<>(new SimpleStringSchema()), props,
FlinkKafkaProducer011.Semantic.EXACTLY_ONCE));

```
In `Flink` version `1.11` (which I'm using with Java), the `SimpleStringSchema` needs a wrapper (ie. `KeyedSerializationSchemaWrapper`) which is also used by @Ankit but removed from the suggested solution as I was getting below `constructor` related error due to the same.

```
FlinkKafkaProducer<String> producer = new FlinkKafkaProducer<String>(
                        topic_name, new KeyedSerializationSchemaWrapper<>(new SimpleStringSchema()), 
                        properties, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);

```

Error:

```
The constructor FlinkKafkaProducer<String>(String, SimpleStringSchema, Properties, FlinkKafkaProducer.Semantic) is undefined

```