Pfthb

I tried the code below and I searched a lot from google, but still can't find my answer.

I can you help me please?

This is my source code:

public static void main( String args ) throws IOException

 String fileName = "path\src\Samp-le\Data1.json";

 try (Stream<String> Data = Files.lines( Paths.get( fileName ) ))
 
 Stream<String> results = Data.Stream < String > ( )
 .filter( str -> str.length() > 6 )// filter the stream to create a new stream
 .collect( Collectors.toList() );
 results.forEach( System.out::println );
 

And here I want to post the json data into kafka producer.

String topicName="test-topic";
String key="sample key";
String value=Data.toString();
Properties prop=new Properties();
prop.put("bootstrap.servers", "localhost:9092,localhost:9093");
prop.put("key.serializer", 
 "org.apache.kafka.common.serialization.StringSerializer");
prop.put("value.serializer", 
 "org.apache.kafka.common.serialization.StringSerializer");
KafkaProducer<String,String> sampleProducer=new KafkaProducer<String,String>(prop);
ProducerRecord<String,String>record=new ProducerRecord<String,String>(topicName,value.toString());
sampleProducer.send(record);
sampleProducer.close();

Following is the json.

 
 "employee": 
 "firstName":"Lokesh",
 "lastName":"Gupta",
 "website":"howtodoinjava.com"
 
,
 
 "employee": 
 "firstName":"Brian",
 "lastName":"Schultz",
 "website":"example.com"
 

If you want to produce to Kafka, replace System.out::println with producer.send(record) after you map a line into a ProducerRecord

There should be no difference in sending a single string message and what you're trying to do. Only the source of your data is different, and you happen to be filtering it on some arbitrary conditions... (rather than length, you should make sure the JSON is actually valid and contains the information you need in some consumer)

Note: If you have more than one partition in your Kafka topic, you'll lose consumption ordering with your current code because you're using a null key in your messages