Hi,
I am using below code to read data from AWS Kinesis stream. But it is
giving me the request body and not the request header. How to get the
request header from Kinesis. My flink jar versions are:
flink-java - 1.6.1
flink-streaming-java_2.11 - 1.6.1
flink-connector-kinesis_2.11 - 1.6.1
My code is:
Properties consumerConfig = new Properties();
consumerConfig.put(ConsumerConfigConstants.AWS_REGION, "<Region>");
consumerConfig.put(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "<KEY>");
consumerConfig.put(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY,
"<SECRET>");
consumerConfig.put(ConsumerConfigConstants.STREAM_INITIAL_POSITION,
"LATEST");
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<String> kinesis = env.addSource(
new FlinkKinesisConsumer<String>("<STREAM>", new SimpleStringSchema(),
consumerConfig));
kinesis.addSink(new SinkFunction<String>() {
@Override
public void invoke(String value) throws Exception {
System.out.println("value= "+value);
}
});
env.execute();
Thanks and Regards,
Madhusudan B. Shastri
