[FLINK-1569] Object reuse mode is not working with KeySelector functions.

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 0.9
• Fix Version/s: 0.9
• Assignee: Aljoscha Krettek
• Reporter: Robert Metzger
• Created: 17/Feb/15 12:49
• Updated: 28/Feb/19 14:29
• Resolved: 02/Mar/15 16:34

Description

The following code works correctly when object reuse is switched off.
When switching it on, the results are wrong.
Using a string-based key selection (putting "name") works for both cases.

	@Test
	public void testWithAvroGenericSer() throws Exception {
		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
	//	env.getConfig().enableObjectReuse();
		Path in = new Path(inFile.getAbsoluteFile().toURI());

		AvroInputFormat<User> users = new AvroInputFormat<User>(in, User.class);
		DataSet<User> usersDS = env.createInput(users);

		DataSet<Tuple2<String, Integer>> res = usersDS.groupBy(new KeySelector<User, String>() {
			@Override
			public String getKey(User value) throws Exception {
				return String.valueOf(value.getName());
			}
		}).reduceGroup(new GroupReduceFunction<User, Tuple2<String, Integer>>() {
			@Override
			public void reduce(Iterable<User> values, Collector<Tuple2<String, Integer>> out) throws Exception {
				for(User u : values) {
					out.collect(new Tuple2<String, Integer>(u.getName().toString(), 1));
				}
			}
		});

		res.writeAsText(resultPath);
		res.print();
		env.execute("Avro Key selection");


		expected = "(Charlie,1)\n(Alyssa,1)\n";
	}