adding source not serializable exception in streaming implementation

[subash basnet]

Hello all,

My requirement is to re-read the csv file from a file path at certain time
intervals and process the csv data. The csv file gets updated at regular
intervals.
Below is my code:
StreamExecutionEnvironment see =
StreamExecutionEnvironment.getExecutionEnvironment();
*DataStream<String> dataStream = getCsvDataStream(see);*
DataStream<Stock> edits = see.addSource(new FetchStock("path/to/csv"));

In FetchStock.java
public class FetchStock extends RichSourceFunction<Stock> {
public FetchStock(String csvPath) {
this.csvPath = csvPath;
}
}

I am trying to adapt code from *WikipediaAnalysis, *but getting the below
not serializable exception on adding source:
*Exception in thread "main"
org.apache.flink.api.common.InvalidProgramException: Object
wikiedits.FetchStock@d7b1517 not serializable*
at
org.apache.flink.api.java.ClosureCleaner.ensureSerializable(ClosureCleaner.java:99)
at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:61)
at
org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.clean(StreamExecutionEnvironment.java:1219)
at
org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.addSource(StreamExecutionEnvironment.java:1131)
at
org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.addSource(StreamExecutionEnvironment.java:1075)
at
org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.addSource(StreamExecutionEnvironment.java:1057)
at wikiedits.StockAnalysis.main(StockAnalysis.java:30)
*Caused by: java.io.NotSerializableException:
org.apache.flink.streaming.api.environment.LocalStreamEnvironment*
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184)
at
java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
at
java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
at
org.apache.flink.util.InstantiationUtil.serializeObject(InstantiationUtil.java:300)
at
org.apache.flink.api.java.ClosureCleaner.ensureSerializable(ClosureCleaner.java:97)
... 6 more


I have attached Stock.java which is just a model with getters and setters.
Not sure what am I doing wrong.

Best Regards,
Subash Basnet

package wikiedits;

import java.io.Serializable;

public class Stock{

	// private String symbol;
	// private double price;
	// private int volume;
	// private double pe;
	// private double eps;
	// private double week52low;
	// private double week52high;
	// private double daylow;
	// private double dayhigh;
	// private double movingav50day;
	// private double marketcap;
	// private String name;
	// private String currency;
	// private double shortRatio;
	// private double previousClose;
	// private double open;
	// private String exchange;

	private String id;
	private int timeStamp;
	private double close;
	private double high;
	private double low;
	private double open;
	private int volume;

	public Stock() {
	}

	public Stock(int timeStamp, double close, double high, double low, double open, int volume, String id) {
		this.timeStamp = timeStamp;
		this.close = close;
		this.high = high;
		this.low = low;
		this.open = open;
		this.volume = volume;
		this.id = id;
	}

	public String getId() {
		return id;
	}

	public void setId(String id) {
		this.id = id;
	}

	public double getLow() {
		return low;
	}

	public void setLow(double low) {
		this.low = low;
	}

	public int getTimeStamp() {
		return timeStamp;
	}

	public void setTimeStamp(int timeStamp) {
		this.timeStamp = timeStamp;
	}

	public double getClose() {
		return close;
	}

	public void setClose(double close) {
		this.close = close;
	}

	public double getHigh() {
		return high;
	}

	public void setHigh(double high) {
		this.high = high;
	}

	public double getOpen() {
		return open;
	}

	public void setOpen(double open) {
		this.open = open;
	}

	public int getVolume() {
		return volume;
	}

	public void setVolume(int volume) {
		this.volume = volume;
	}

}