Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[FLINK-16307][e2e] Use start-cluster.sh instead of manually building cluster #11246

Merged
merged 2 commits into from Feb 28, 2020
Merged

[FLINK-16307][e2e] Use start-cluster.sh instead of manually building cluster #11246

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
446c719
[FLINK-16307][e2e] Use start-cluster.sh instead of manually building …
zentol Feb 27, 2020
db7d7f5
+
zentol Feb 28, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
5 changes: 5 additions & 0 deletions ...-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/FlinkDistribution.java
View file
Open in desktop
Expand Up @@ -48,6 +48,7 @@
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Optional;
Expand Down Expand Up @@ -288,6 +289,10 @@ public void appendConfiguration(Configuration config) throws IOException {
Files.write(conf.resolve("flink-conf.yaml"), configurationLines);
}

public void setTaskExecutorHosts(Collection<String> taskExecutorHosts) throws IOException {
Files.write(conf.resolve("slaves"), taskExecutorHosts);
}

public Stream<String> searchAllLogs(Pattern pattern, Function<Matcher, String> matchProcessor) throws IOException {
final List<String> matches = new ArrayList<>(2);

Expand Down
8 changes: 4 additions & 4 deletions ...-common/src/main/java/org/apache/flink/tests/util/flink/LocalStandaloneFlinkResource.java
View file
Open in desktop
Expand Up @@ -39,6 +39,8 @@
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
* Flink resource that start local standalone clusters.
Expand Down Expand Up @@ -71,10 +73,8 @@ public void addConfiguration(final Configuration config) throws IOException {

@Override
public ClusterController startCluster(int numTaskManagers) throws IOException {
distribution.startJobManager();
for (int x = 0; x < numTaskManagers; x++) {
distribution.startTaskManager();
}
distribution.setTaskExecutorHosts(IntStream.range(0, numTaskManagers).mapToObj(i -> "localhost").collect(Collectors.toList()));
zentol marked this conversation as resolved.
Show resolved Hide resolved
distribution.startFlinkCluster();

try (final RestClient restClient = new RestClient(RestClientConfiguration.fromConfiguration(new Configuration()), Executors.directExecutor())) {
for (int retryAttempt = 0; retryAttempt < 30; retryAttempt++) {
Expand Down