Abacn commented on code in PR #39272:
URL: https://github.com/apache/beam/pull/39272#discussion_r3559533786
##########
runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkSubmissionTest.java:
##########
@@ -230,21 +231,50 @@ private static void restoreEnvironment() throws Exception
{
* We modify the JVM's environment variables here. This is necessary for the
end-to-end test
* because Flink's CliFrontend requires a Flink configuration file for which
the location can only
* be set using the {@code ConfigConstants.ENV_FLINK_CONF_DIR} environment
variable.
+ *
+ * <p>On Unix JDKs, {@code ProcessEnvironment.theEnvironment} is a {@code
Map<Variable,Value>}.
+ * {@code System.getenv(String)} looks up {@code Variable} keys, so
inserting plain {@code String}
+ * keys (via {@code putAll}) is a no-op for lookups and leaves Flink unable
to find {@code
+ * FLINK_CONF_DIR}. On Windows the map uses {@code String} keys (plus a
case-insensitive map).
*/
+ @SuppressWarnings("unchecked")
private static void modifyEnv(Map<String, String> env) throws Exception {
- Class processEnv = Class.forName("java.lang.ProcessEnvironment");
- Field envField = processEnv.getDeclaredField("theUnmodifiableEnvironment");
+ Class<?> processEnvironment =
Class.forName("java.lang.ProcessEnvironment");
- Field modifiersField = Field.class.getDeclaredField("modifiers");
- modifiersField.setAccessible(true);
- modifiersField.setInt(envField, envField.getModifiers() & ~Modifier.FINAL);
-
- envField.setAccessible(true);
- envField.set(null, env);
- envField.setAccessible(false);
+ Field theEnvironmentField =
processEnvironment.getDeclaredField("theEnvironment");
+ theEnvironmentField.setAccessible(true);
+ Map<Object, Object> envMap = (Map<Object, Object>)
theEnvironmentField.get(null);
+ envMap.clear();
+ try {
+ // Unix: keys/values are ProcessEnvironment$Variable / $Value, not
String.
+ Class<?> variableClass =
Class.forName("java.lang.ProcessEnvironment$Variable");
+ Class<?> valueClass =
Class.forName("java.lang.ProcessEnvironment$Value");
+ Method valueOfVariable = variableClass.getDeclaredMethod("valueOf",
String.class);
+ Method valueOfValue = valueClass.getDeclaredMethod("valueOf",
String.class);
+ valueOfVariable.setAccessible(true);
+ valueOfValue.setAccessible(true);
+ for (Map.Entry<String, String> entry : env.entrySet()) {
+ envMap.put(
Review Comment:
Please minimize the scope of try { ... }. I understand it should only
affects Class.forName.
##########
runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkSubmissionTest.java:
##########
@@ -230,21 +231,50 @@ private static void restoreEnvironment() throws Exception
{
* We modify the JVM's environment variables here. This is necessary for the
end-to-end test
* because Flink's CliFrontend requires a Flink configuration file for which
the location can only
* be set using the {@code ConfigConstants.ENV_FLINK_CONF_DIR} environment
variable.
+ *
+ * <p>On Unix JDKs, {@code ProcessEnvironment.theEnvironment} is a {@code
Map<Variable,Value>}.
+ * {@code System.getenv(String)} looks up {@code Variable} keys, so
inserting plain {@code String}
+ * keys (via {@code putAll}) is a no-op for lookups and leaves Flink unable
to find {@code
+ * FLINK_CONF_DIR}. On Windows the map uses {@code String} keys (plus a
case-insensitive map).
*/
+ @SuppressWarnings("unchecked")
private static void modifyEnv(Map<String, String> env) throws Exception {
- Class processEnv = Class.forName("java.lang.ProcessEnvironment");
- Field envField = processEnv.getDeclaredField("theUnmodifiableEnvironment");
+ Class<?> processEnvironment =
Class.forName("java.lang.ProcessEnvironment");
- Field modifiersField = Field.class.getDeclaredField("modifiers");
- modifiersField.setAccessible(true);
- modifiersField.setInt(envField, envField.getModifiers() & ~Modifier.FINAL);
-
- envField.setAccessible(true);
- envField.set(null, env);
- envField.setAccessible(false);
+ Field theEnvironmentField =
processEnvironment.getDeclaredField("theEnvironment");
Review Comment:
nit: there is redundant diff (variable renamed to verbose). One should focus
on test fix itself
##########
runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/UnboundedSourceWrapperTest.java:
##########
@@ -665,15 +665,16 @@ private static void testSourceDoesNotShutdown(boolean
shouldHaveReaders) throws
// Wait to see if the wrapper shuts down immediately in case it
doesn't have readers
if (!shouldHaveReaders) {
// The expected state is for finalizeSource to sleep instead of
exiting
- while (true) {
- StackTraceElement[] callStack = thread.getStackTrace();
- if (callStack.length >= 2
- && "sleep".equals(callStack[0].getMethodName())
- && "finalizeSource".equals(callStack[1].getMethodName())) {
+ long deadlineNs = System.nanoTime() + 9_000_000_000L;
+ boolean reachedFinalizeSource = false;
+ while (System.nanoTime() < deadlineNs) {
Review Comment:
Do not changing while (true) to an arbitrary large "System.nanoTime() +
9_000_000_000L".
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
