You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "tisonkun (via GitHub)" <gi...@apache.org> on 2023/07/18 08:32:53 UTC
[GitHub] [flink-connector-pulsar] tisonkun opened a new pull request, #55: [FLINK-24302] Test coverage for JDK 11
tisonkun opened a new pull request, #55:
URL: https://github.com/apache/flink-connector-pulsar/pull/55
## Purpose of the change
*For example: Add dynamic sink topic support for Pulsar connector.*
## Brief change log
- *Change the internal design of `ProducerRegister`.*
- *Expose topic metadata query in `PulsarSinkContext`.*
- *Change the internal metadata cache in `MetadataListener`.*
## Verifying this change
This change is already covered by existing tests, such as *(please describe tests)*.
## Significant changes
*(Please check any boxes [x] if the answer is "yes". You can first publish the PR and check them afterwards, for
convenience.)*
- [ ] Dependencies have been added or upgraded
- [ ] Public API has been changed (Public API is any class annotated with `@Public(Evolving)`)
- [ ] Serializers have been changed
- [ ] New feature has been introduced
- If yes, how is this documented? (not applicable / docs / JavaDocs / not documented)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-pulsar] tisonkun commented on pull request #55: [FLINK-24302] Extend offheap memory for JDK11 test coverage
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on PR #55:
URL: https://github.com/apache/flink-connector-pulsar/pull/55#issuecomment-1643664256
Merging...
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-pulsar] reswqa commented on a diff in pull request #55: [FLINK-24302] Extend offheap memory for JDK11 test coverage
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on code in PR #55:
URL: https://github.com/apache/flink-connector-pulsar/pull/55#discussion_r1268492288
##########
flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerUtils.java:
##########
@@ -35,9 +35,11 @@ public class FlinkContainerUtils {
public static Configuration flinkConfiguration() {
Configuration configuration = new Configuration();
- // Increase the jvm metaspace memory to avoid java.lang.OutOfMemoryError: Metaspace
- configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(2048));
+ configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(2560));
+ configuration.set(TaskManagerOptions.TASK_OFF_HEAP_MEMORY, MemorySize.ofMebiBytes(512));
configuration.set(TaskManagerOptions.JVM_METASPACE, MemorySize.ofMebiBytes(512));
+
+ // Increase the jvm metaspace memory to avoid java.lang.OutOfMemoryError: Metaspace
configuration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(2560));
configuration.set(JobManagerOptions.JVM_METASPACE, MemorySize.ofMebiBytes(1024));
Review Comment:
Is this more appropriate(perhaps)? But it doesn't matter if we don't make any change. 😆
##########
flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerUtils.java:
##########
@@ -35,9 +35,11 @@ public class FlinkContainerUtils {
public static Configuration flinkConfiguration() {
Configuration configuration = new Configuration();
- // Increase the jvm metaspace memory to avoid java.lang.OutOfMemoryError: Metaspace
- configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(2048));
+ configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(2560));
+ configuration.set(TaskManagerOptions.TASK_OFF_HEAP_MEMORY, MemorySize.ofMebiBytes(512));
configuration.set(TaskManagerOptions.JVM_METASPACE, MemorySize.ofMebiBytes(512));
+
+ // Increase the jvm metaspace memory to avoid java.lang.OutOfMemoryError: Metaspace
configuration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(2560));
configuration.set(JobManagerOptions.JVM_METASPACE, MemorySize.ofMebiBytes(1024));
Review Comment:
Is this more appropriate(perhaps)? But it doesn't matter if we don't make any change. 😆
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-pulsar] tisonkun commented on pull request #55: [FLINK-24302] Extend offheap memory for JDK11 test coverage
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on PR #55:
URL: https://github.com/apache/flink-connector-pulsar/pull/55#issuecomment-1643457403
Apply suggestions.
Pending to merge when CI reports green.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-pulsar] tisonkun commented on a diff in pull request #55: [FLINK-24302] Extend offheap memory for JDK11 test coverage
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on code in PR #55:
URL: https://github.com/apache/flink-connector-pulsar/pull/55#discussion_r1268323229
##########
flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/MiniClusterTestEnvironment.java:
##########
@@ -0,0 +1,184 @@
+/*
+ * 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.connector.pulsar.common;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.testframe.environment.ClusterControllable;
+import org.apache.flink.connector.testframe.environment.TestEnvironment;
+import org.apache.flink.connector.testframe.environment.TestEnvironmentSettings;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.runtime.highavailability.nonha.embedded.HaLeadershipControl;
+import org.apache.flink.runtime.minicluster.RpcServiceSharing;
+import org.apache.flink.runtime.taskexecutor.TaskExecutorResourceUtils;
+import org.apache.flink.runtime.testutils.CommonTestUtils;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.util.TestStreamEnvironment;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
+
+import org.apache.commons.io.FileUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.configuration.MetricOptions.METRIC_FETCHER_UPDATE_INTERVAL;
+import static org.apache.flink.connector.testframe.utils.ConnectorTestConstants.METRIC_FETCHER_UPDATE_INTERVAL_MS;
+import static org.apache.flink.runtime.jobgraph.SavepointConfigOptions.SAVEPOINT_PATH;
+
+/** Test environment for running jobs on Flink mini-cluster. */
+@Experimental
+public class MiniClusterTestEnvironment implements TestEnvironment, ClusterControllable {
Review Comment:
Yes and when we drop the support at least test coverage for 1.17.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-pulsar] syhily commented on pull request #55: [FLINK-24302] Test coverage for JDK 11
Posted by "syhily (via GitHub)" <gi...@apache.org>.
syhily commented on PR #55:
URL: https://github.com/apache/flink-connector-pulsar/pull/55#issuecomment-1639890230
Too sad to see that the memory leak issue on Java 11 still exists on Pulsar 3.0.0 client.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-pulsar] reswqa commented on a diff in pull request #55: [FLINK-24302] Extend offheap memory for JDK11 test coverage
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on code in PR #55:
URL: https://github.com/apache/flink-connector-pulsar/pull/55#discussion_r1268484338
##########
flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerUtils.java:
##########
@@ -35,9 +35,11 @@ public class FlinkContainerUtils {
public static Configuration flinkConfiguration() {
Configuration configuration = new Configuration();
- // Increase the jvm metaspace memory to avoid java.lang.OutOfMemoryError: Metaspace
- configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(2048));
+ configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(2560));
+ configuration.set(TaskManagerOptions.TASK_OFF_HEAP_MEMORY, MemorySize.ofMebiBytes(512));
configuration.set(TaskManagerOptions.JVM_METASPACE, MemorySize.ofMebiBytes(512));
+
+ // Increase the jvm metaspace memory to avoid java.lang.OutOfMemoryError: Metaspace
configuration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(2560));
configuration.set(JobManagerOptions.JVM_METASPACE, MemorySize.ofMebiBytes(1024));
Review Comment:
```suggestion
configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(2560));
configuration.set(TaskManagerOptions.TASK_OFF_HEAP_MEMORY, MemorySize.ofMebiBytes(512));
// Increase the jvm metaspace memory to avoid java.lang.OutOfMemoryError: Metaspace
configuration.set(TaskManagerOptions.JVM_METASPACE, MemorySize.ofMebiBytes(512));
configuration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(2560));
// Increase the jvm metaspace memory to avoid java.lang.OutOfMemoryError: Metaspace
configuration.set(JobManagerOptions.JVM_METASPACE, MemorySize.ofMebiBytes(1024));
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-pulsar] tisonkun merged pull request #55: [FLINK-24302] Extend offheap memory for JDK11 test coverage
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun merged PR #55:
URL: https://github.com/apache/flink-connector-pulsar/pull/55
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-pulsar] syhily commented on pull request #55: [FLINK-24302] Extend offheap memory for JDK11 test coverage
Posted by "syhily (via GitHub)" <gi...@apache.org>.
syhily commented on PR #55:
URL: https://github.com/apache/flink-connector-pulsar/pull/55#issuecomment-1642986077
Cool, I'm so happy to see that we finally have the JDK 11 test support. The extensive off heap memory usage in Pulsar client could be the main cause of the OOM in JDK 11. Should we mention it in our document if the user want to use this connector in JDK 11 or above? @reswqa @tisonkun
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-pulsar] reswqa commented on a diff in pull request #55: [FLINK-24302] Extend offheap memory for JDK11 test coverage
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on code in PR #55:
URL: https://github.com/apache/flink-connector-pulsar/pull/55#discussion_r1268105952
##########
flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/common/MiniClusterTestEnvironment.java:
##########
@@ -0,0 +1,184 @@
+/*
+ * 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.connector.pulsar.common;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.testframe.environment.ClusterControllable;
+import org.apache.flink.connector.testframe.environment.TestEnvironment;
+import org.apache.flink.connector.testframe.environment.TestEnvironmentSettings;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.runtime.highavailability.nonha.embedded.HaLeadershipControl;
+import org.apache.flink.runtime.minicluster.RpcServiceSharing;
+import org.apache.flink.runtime.taskexecutor.TaskExecutorResourceUtils;
+import org.apache.flink.runtime.testutils.CommonTestUtils;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.util.TestStreamEnvironment;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
+
+import org.apache.commons.io.FileUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.configuration.MetricOptions.METRIC_FETCHER_UPDATE_INTERVAL;
+import static org.apache.flink.connector.testframe.utils.ConnectorTestConstants.METRIC_FETCHER_UPDATE_INTERVAL_MS;
+import static org.apache.flink.runtime.jobgraph.SavepointConfigOptions.SAVEPOINT_PATH;
+
+/** Test environment for running jobs on Flink mini-cluster. */
+@Experimental
+public class MiniClusterTestEnvironment implements TestEnvironment, ClusterControllable {
Review Comment:
IIUC, we will remove this after `FLINK-1.18` released?
##########
flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerUtils.java:
##########
@@ -35,9 +35,11 @@ public class FlinkContainerUtils {
public static Configuration flinkConfiguration() {
Configuration configuration = new Configuration();
- // Increase the jvm metaspace memory to avoid java.lang.OutOfMemoryError: Metaspace
- configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(2048));
+ configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(2560));
+ configuration.set(TaskManagerOptions.TASK_OFF_HEAP_MEMORY, MemorySize.ofMebiBytes(512));
configuration.set(TaskManagerOptions.JVM_METASPACE, MemorySize.ofMebiBytes(512));
+
+ // Increase the jvm metaspace memory to avoid java.lang.OutOfMemoryError: Metaspace
Review Comment:
Should this line of comment be moved forward a bit?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-pulsar] tisonkun commented on a diff in pull request #55: [FLINK-24302] Extend offheap memory for JDK11 test coverage
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on code in PR #55:
URL: https://github.com/apache/flink-connector-pulsar/pull/55#discussion_r1268324359
##########
flink-connector-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/common/FlinkContainerUtils.java:
##########
@@ -35,9 +35,11 @@ public class FlinkContainerUtils {
public static Configuration flinkConfiguration() {
Configuration configuration = new Configuration();
- // Increase the jvm metaspace memory to avoid java.lang.OutOfMemoryError: Metaspace
- configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(2048));
+ configuration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(2560));
+ configuration.set(TaskManagerOptions.TASK_OFF_HEAP_MEMORY, MemorySize.ofMebiBytes(512));
configuration.set(TaskManagerOptions.JVM_METASPACE, MemorySize.ofMebiBytes(512));
+
+ // Increase the jvm metaspace memory to avoid java.lang.OutOfMemoryError: Metaspace
Review Comment:
Please directly comment a suggestion or patch. I'm not sure where to move in your mind,
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-pulsar] tisonkun commented on pull request #55: [FLINK-24302] Test coverage for JDK 11
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on PR #55:
URL: https://github.com/apache/flink-connector-pulsar/pull/55#issuecomment-1641787804
Pending to merge...
@syhily do we have a ticket for supporting table connector already? Or I should open a new one? It should be my next step to track.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org