<commit_before>/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.flink.api.common.typeutils;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.memory.DataInputDeserializer;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputSerializer;
import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.testutils.migration.MigrationVersion;
import org.apache.flink.util.TestLogger;

import org.hamcrest.Matcher;
import org.junit.Ignore;
import org.junit.Test;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Collections;
import java.util.List;

import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.hamcrest.CoreMatchers.not;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import static org.junit.Assume.assumeThat;

/**
 * A test base for testing {@link TypeSerializer} upgrades.
 * <p>
 * You can run {@link #generateTestSetupFiles()} on a Flink branch to (re-)generate the test data
 * files.
 */
public abstract class TypeSerializerUpgradeTestBase<PreviousElementT, UpgradedElementT> extends TestLogger {

	protected static final MigrationVersion[] migrationVersions = new MigrationVersion[]{
			MigrationVersion.v1_7,
			MigrationVersion.v1_8,
			MigrationVersion.v1_9,
	};

	private final TestSpecification<PreviousElementT, UpgradedElementT> testSpecification;

	protected TypeSerializerUpgradeTestBase(TestSpecification<PreviousElementT, UpgradedElementT> testSpecification) {
		this.testSpecification = checkNotNull(testSpecification);
	}

	// ------------------------------------------------------------------------------
	//  APIs
	// ------------------------------------------------------------------------------

	public interface PreUpgradeSetup<PreviousElementT> {
		TypeSerializer<PreviousElementT> createPriorSerializer();
		PreviousElementT createTestData();
	}

	public interface UpgradeVerifier<UpgradedElementT> {
		TypeSerializer<UpgradedElementT> createUpgradedSerializer();
		UpgradedElementT expectedTestData();
		Matcher<TypeSerializerSchemaCompatibility<UpgradedElementT>> schemaCompatibilityMatcher();
	}

	private static class ClassLoaderSafePreUpgradeSetup<PreviousElementT> implements PreUpgradeSetup<PreviousElementT> {

		private final PreUpgradeSetup<PreviousElementT> delegateSetup;
		private final ClassLoader setupClassloader;

		ClassLoaderSafePreUpgradeSetup(Class<? extends PreUpgradeSetup<PreviousElementT>> delegateSetupClass) throws Exception {
			checkNotNull(delegateSetupClass);
			Class<? extends PreUpgradeSetup<PreviousElementT>> relocatedDelegateSetupClass =
				ClassRelocator.relocate(delegateSetupClass);

			this.setupClassloader = relocatedDelegateSetupClass.getClassLoader();
			try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(setupClassloader)) {
				this.delegateSetup = relocatedDelegateSetupClass.newInstance();
			}
		}

		@Override
		public TypeSerializer<PreviousElementT> createPriorSerializer() {
			try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(setupClassloader)) {
				return delegateSetup.createPriorSerializer();
			} catch (IOException e) {
				throw new RuntimeException("Error creating prior serializer via ClassLoaderSafePreUpgradeSetup.", e);
			}
		}

		@Override
		public PreviousElementT createTestData() {
			try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(setupClassloader)) {
				return delegateSetup.createTestData();
			} catch (IOException e) {
				throw new RuntimeException("Error creating test data via ThreadContextClassLoader.", e);
			}
		}
	}

	private static class ClassLoaderSafeUpgradeVerifier<UpgradedElementT> implements UpgradeVerifier<UpgradedElementT> {

		private final UpgradeVerifier<UpgradedElementT> delegateVerifier;
		private final ClassLoader verifierClassloader;

		ClassLoaderSafeUpgradeVerifier(Class<? extends UpgradeVerifier<UpgradedElementT>> delegateVerifierClass) throws Exception {
			checkNotNull(delegateVerifierClass);
			Class<? extends UpgradeVerifier<UpgradedElementT>> relocatedDelegateVerifierClass =
				ClassRelocator.relocate(delegateVerifierClass);

			this.verifierClassloader = relocatedDelegateVerifierClass.getClassLoader();
			try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(verifierClassloader)) {
				this.delegateVerifier = relocatedDelegateVerifierClass.newInstance();
			}
		}

		@Override
		public TypeSerializer<UpgradedElementT> createUpgradedSerializer() {
			try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(verifierClassloader)) {
				return delegateVerifier.createUpgradedSerializer();
			} catch (IOException e) {
				throw new RuntimeException("Error creating upgraded serializer via ClassLoaderSafeUpgradeVerifier.", e);
			}
		}

		@Override
		public UpgradedElementT expectedTestData() {
			try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(verifierClassloader)) {
				return delegateVerifier.expectedTestData();
			} catch (IOException e) {
				throw new RuntimeException("Error creating expected test data via ClassLoaderSafeUpgradeVerifier.", e);
			}
		}

		@Override
		public Matcher<TypeSerializerSchemaCompatibility<UpgradedElementT>> schemaCompatibilityMatcher() {
			try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(verifierClassloader)) {
				return delegateVerifier.schemaCompatibilityMatcher();
			} catch (IOException e) {
				throw new RuntimeException("Error creating schema compatibility matcher via ClassLoaderSafeUpgradeVerifier.", e);
			}
		}
	}

	public static class TestSpecification<PreviousElementT, UpgradedElementT> {
		private final String name;
		private final MigrationVersion migrationVersion;
		private final ClassLoaderSafePreUpgradeSetup<PreviousElementT> setup;
		private final ClassLoaderSafeUpgradeVerifier<UpgradedElementT> verifier;

		public TestSpecification(
				String name,
				MigrationVersion migrationVersion,
				Class<? extends PreUpgradeSetup<PreviousElementT>> setupClass,
				Class<? extends UpgradeVerifier<UpgradedElementT>> verifierClass) throws Exception {
			this.name = checkNotNull(name);
			this.migrationVersion = checkNotNull(migrationVersion);
			this.setup = new ClassLoaderSafePreUpgradeSetup<>(setupClass);
			this.verifier = new ClassLoaderSafeUpgradeVerifier<>(verifierClass);
		}

		@Override
		public String toString() {
			return name + " / " + migrationVersion;
		}
	}

	// ------------------------------------------------------------------------------
	//  Test file generation
	// ------------------------------------------------------------------------------

	private static final int INITIAL_OUTPUT_BUFFER_SIZE = 64;

	/**
	 * Execute this test to generate test files.
	 * Remember to be using the correct branch when generating the test files,
	 * e.g. to generate test files for {@link MigrationVersion#v1_8}, you should be under the
	 * release-1.8 branch.
	 */
	@Ignore
	@Test
	public void generateTestSetupFiles() throws Exception {
		Files.createDirectory(getSerializerSnapshotFilePath().getParent());

		try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(testSpecification.setup.setupClassloader)) {
			TypeSerializer<PreviousElementT> priorSerializer = testSpecification.setup.createPriorSerializer();

			// first, use the serializer to write test data
			// NOTE: it is important that we write test data first, because some serializers' configuration
			//       mutates only after being used for serialization (e.g. dynamic type registrations for Pojo / Kryo)
			DataOutputSerializer testDataOut = new DataOutputSerializer(INITIAL_OUTPUT_BUFFER_SIZE);
			priorSerializer.serialize(testSpecification.setup.createTestData(), testDataOut);<commit_msg>[hotfix] Fix directory creation in TypeSerializerUpgradeTestBase<commit_after>/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.flink.api.common.typeutils;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.memory.DataInputDeserializer;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputSerializer;
import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.testutils.migration.MigrationVersion;
import org.apache.flink.util.TestLogger;

import org.hamcrest.Matcher;
import org.junit.Ignore;
import org.junit.Test;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Collections;
import java.util.List;

import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.hamcrest.CoreMatchers.not;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import static org.junit.Assume.assumeThat;

/**
 * A test base for testing {@link TypeSerializer} upgrades.
 * <p>
 * You can run {@link #generateTestSetupFiles()} on a Flink branch to (re-)generate the test data
 * files.
 */
public abstract class TypeSerializerUpgradeTestBase<PreviousElementT, UpgradedElementT> extends TestLogger {

	protected static final MigrationVersion[] migrationVersions = new MigrationVersion[]{
			MigrationVersion.v1_7,
			MigrationVersion.v1_8,
			MigrationVersion.v1_9,
	};

	private final TestSpecification<PreviousElementT, UpgradedElementT> testSpecification;

	protected TypeSerializerUpgradeTestBase(TestSpecification<PreviousElementT, UpgradedElementT> testSpecification) {
		this.testSpecification = checkNotNull(testSpecification);
	}

	// ------------------------------------------------------------------------------
	//  APIs
	// ------------------------------------------------------------------------------

	public interface PreUpgradeSetup<PreviousElementT> {
		TypeSerializer<PreviousElementT> createPriorSerializer();
		PreviousElementT createTestData();
	}

	public interface UpgradeVerifier<UpgradedElementT> {
		TypeSerializer<UpgradedElementT> createUpgradedSerializer();
		UpgradedElementT expectedTestData();
		Matcher<TypeSerializerSchemaCompatibility<UpgradedElementT>> schemaCompatibilityMatcher();
	}

	private static class ClassLoaderSafePreUpgradeSetup<PreviousElementT> implements PreUpgradeSetup<PreviousElementT> {

		private final PreUpgradeSetup<PreviousElementT> delegateSetup;
		private final ClassLoader setupClassloader;

		ClassLoaderSafePreUpgradeSetup(Class<? extends PreUpgradeSetup<PreviousElementT>> delegateSetupClass) throws Exception {
			checkNotNull(delegateSetupClass);
			Class<? extends PreUpgradeSetup<PreviousElementT>> relocatedDelegateSetupClass =
				ClassRelocator.relocate(delegateSetupClass);

			this.setupClassloader = relocatedDelegateSetupClass.getClassLoader();
			try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(setupClassloader)) {
				this.delegateSetup = relocatedDelegateSetupClass.newInstance();
			}
		}

		@Override
		public TypeSerializer<PreviousElementT> createPriorSerializer() {
			try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(setupClassloader)) {
				return delegateSetup.createPriorSerializer();
			} catch (IOException e) {
				throw new RuntimeException("Error creating prior serializer via ClassLoaderSafePreUpgradeSetup.", e);
			}
		}

		@Override
		public PreviousElementT createTestData() {
			try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(setupClassloader)) {
				return delegateSetup.createTestData();
			} catch (IOException e) {
				throw new RuntimeException("Error creating test data via ThreadContextClassLoader.", e);
			}
		}
	}

	private static class ClassLoaderSafeUpgradeVerifier<UpgradedElementT> implements UpgradeVerifier<UpgradedElementT> {

		private final UpgradeVerifier<UpgradedElementT> delegateVerifier;
		private final ClassLoader verifierClassloader;

		ClassLoaderSafeUpgradeVerifier(Class<? extends UpgradeVerifier<UpgradedElementT>> delegateVerifierClass) throws Exception {
			checkNotNull(delegateVerifierClass);
			Class<? extends UpgradeVerifier<UpgradedElementT>> relocatedDelegateVerifierClass =
				ClassRelocator.relocate(delegateVerifierClass);

			this.verifierClassloader = relocatedDelegateVerifierClass.getClassLoader();
			try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(verifierClassloader)) {
				this.delegateVerifier = relocatedDelegateVerifierClass.newInstance();
			}
		}

		@Override
		public TypeSerializer<UpgradedElementT> createUpgradedSerializer() {
			try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(verifierClassloader)) {
				return delegateVerifier.createUpgradedSerializer();
			} catch (IOException e) {
				throw new RuntimeException("Error creating upgraded serializer via ClassLoaderSafeUpgradeVerifier.", e);
			}
		}

		@Override
		public UpgradedElementT expectedTestData() {
			try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(verifierClassloader)) {
				return delegateVerifier.expectedTestData();
			} catch (IOException e) {
				throw new RuntimeException("Error creating expected test data via ClassLoaderSafeUpgradeVerifier.", e);
			}
		}

		@Override
		public Matcher<TypeSerializerSchemaCompatibility<UpgradedElementT>> schemaCompatibilityMatcher() {
			try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(verifierClassloader)) {
				return delegateVerifier.schemaCompatibilityMatcher();
			} catch (IOException e) {
				throw new RuntimeException("Error creating schema compatibility matcher via ClassLoaderSafeUpgradeVerifier.", e);
			}
		}
	}

	public static class TestSpecification<PreviousElementT, UpgradedElementT> {
		private final String name;
		private final MigrationVersion migrationVersion;
		private final ClassLoaderSafePreUpgradeSetup<PreviousElementT> setup;
		private final ClassLoaderSafeUpgradeVerifier<UpgradedElementT> verifier;

		public TestSpecification(
				String name,
				MigrationVersion migrationVersion,
				Class<? extends PreUpgradeSetup<PreviousElementT>> setupClass,
				Class<? extends UpgradeVerifier<UpgradedElementT>> verifierClass) throws Exception {
			this.name = checkNotNull(name);
			this.migrationVersion = checkNotNull(migrationVersion);
			this.setup = new ClassLoaderSafePreUpgradeSetup<>(setupClass);
			this.verifier = new ClassLoaderSafeUpgradeVerifier<>(verifierClass);
		}

		@Override
		public String toString() {
			return name + " / " + migrationVersion;
		}
	}

	// ------------------------------------------------------------------------------
	//  Test file generation
	// ------------------------------------------------------------------------------

	private static final int INITIAL_OUTPUT_BUFFER_SIZE = 64;

	/**
	 * Execute this test to generate test files.
	 * Remember to be using the correct branch when generating the test files,
	 * e.g. to generate test files for {@link MigrationVersion#v1_8}, you should be under the
	 * release-1.8 branch.
	 */
	@Ignore
	@Test
	public void generateTestSetupFiles() throws Exception {
		Files.createDirectories(getSerializerSnapshotFilePath().getParent());

		try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(testSpecification.setup.setupClassloader)) {
			TypeSerializer<PreviousElementT> priorSerializer = testSpecification.setup.createPriorSerializer();

			// first, use the serializer to write test data
			// NOTE: it is important that we write test data first, because some serializers' configuration
			//       mutates only after being used for serialization (e.g. dynamic type registrations for Pojo / Kryo)
			DataOutputSerializer testDataOut = new DataOutputSerializer(INITIAL_OUTPUT_BUFFER_SIZE);
			priorSerializer.serialize(testSpecification.setup.createTestData(), testDataOut);
{'fraction_non_alphanumeric': 0.11011921902896965, 'fraction_numerical': 0.0019581869492599204, 'mean_word_length': 4.583279742765273, 'pattern_counts': {'":': 0, '<': 79, '<?xml version=': 0, '>': 79, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '835807', 'n_tokens_mistral': 4773, 'n_tokens_neox': 4325, 'n_words': 1150}