This is an automated email from the ASF dual-hosted git repository. orpiske pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/camel-kafka-connector.git
The following commit(s) were added to refs/heads/master by this push: new 2499c8c Align HDFS tests with test infra from Camel 3.7 2499c8c is described below commit 2499c8cee57553d84a2d706e2e716612ccbd559c Author: Otavio Rodolfo Piske <opi...@redhat.com> AuthorDate: Thu Dec 17 09:37:59 2020 +0100 Align HDFS tests with test infra from Camel 3.7 --- tests/itests-hdfs/pom.xml | 17 ++++ .../hdfs/services/ContainerLocalHDFSService.java | 65 -------------- .../hdfs/services/DataNodeContainer.java | 98 ---------------------- .../kafkaconnector/hdfs/services/HDFSPorts.java | 31 ------- .../kafkaconnector/hdfs/services/HDFSService.java | 58 ------------- .../hdfs/services/HDFSServiceFactory.java | 46 ---------- .../hdfs/services/HadoopBaseContainer.java | 48 ----------- .../hdfs/services/NameNodeContainer.java | 50 ----------- .../hdfs/services/RemoteHDFSService.java | 47 ----------- .../hdfs/sink/CamelSinkHDFSITCase.java | 6 +- 10 files changed, 19 insertions(+), 447 deletions(-) diff --git a/tests/itests-hdfs/pom.xml b/tests/itests-hdfs/pom.xml index ed47bd2..4b17a07 100644 --- a/tests/itests-hdfs/pom.xml +++ b/tests/itests-hdfs/pom.xml @@ -37,6 +37,23 @@ <scope>test</scope> </dependency> + <!-- test infra --> + <dependency> + <groupId>org.apache.camel</groupId> + <artifactId>camel-test-infra-common</artifactId> + <version>${camel.version}</version> + <type>test-jar</type> + <scope>test</scope> + </dependency> + + <dependency> + <groupId>org.apache.camel</groupId> + <artifactId>camel-test-infra-hdfs</artifactId> + <version>${camel.version}</version> + <type>test-jar</type> + <scope>test</scope> + </dependency> + <dependency> <groupId>org.apache.camel</groupId> <artifactId>camel-hdfs</artifactId> diff --git a/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/services/ContainerLocalHDFSService.java b/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/services/ContainerLocalHDFSService.java deleted file mode 100644 index ca0947a..0000000 --- a/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/services/ContainerLocalHDFSService.java +++ /dev/null @@ -1,65 +0,0 @@ -/* - * 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.camel.kafkaconnector.hdfs.services; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.testcontainers.containers.Network; - -public class ContainerLocalHDFSService implements HDFSService { - private static final Logger LOG = LoggerFactory.getLogger(ContainerLocalHDFSService.class); - private final NameNodeContainer nameNodeContainer; - private final DataNodeContainer dataNodeContainer; - - public ContainerLocalHDFSService() { - Network network = Network.newNetwork(); - - nameNodeContainer = new NameNodeContainer(network); - dataNodeContainer = new DataNodeContainer(network); - } - - @Override - public String getHDFSHost() { - return nameNodeContainer.getContainerIpAddress(); - } - - @Override - public int getPort() { - return nameNodeContainer.getIpcPort(); - } - - @Override - public void initialize() { - nameNodeContainer.start(); - - String hdfsNameNodeWeb = nameNodeContainer.getContainerIpAddress() + ":" + nameNodeContainer.getHttpPort(); - LOG.info("HDFS Name node web UI running at address http://{}", hdfsNameNodeWeb); - - dataNodeContainer.start(); - - String hdfsDataNodeWeb = dataNodeContainer.getContainerIpAddress() + ":" + dataNodeContainer.getHttpPort(); - LOG.info("HDFS Data node web UI running at address http://{}", hdfsDataNodeWeb); - LOG.info("HDFS Data node running at address {}:{}", getHDFSHost(), getPort()); - } - - @Override - public void shutdown() { - dataNodeContainer.stop(); - nameNodeContainer.stop(); - } -} diff --git a/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/services/DataNodeContainer.java b/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/services/DataNodeContainer.java deleted file mode 100644 index 5cd51e7..0000000 --- a/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/services/DataNodeContainer.java +++ /dev/null @@ -1,98 +0,0 @@ -/* - * 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.camel.kafkaconnector.hdfs.services; - -import java.net.URI; -import java.net.URISyntaxException; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.testcontainers.containers.Network; -import org.testcontainers.containers.wait.strategy.Wait; -import org.testcontainers.dockerclient.DockerClientConfigUtils; - -public class DataNodeContainer extends HadoopBaseContainer<DataNodeContainer> { - private static final Logger LOG = LoggerFactory.getLogger(DataNodeContainer.class); - private static String dataNodeHost = "localhost"; - - static { - String dockerHost = System.getenv("DOCKER_HOST"); - - if (dockerHost != null && !dockerHost.isEmpty()) { - try { - URI dockerHostUri = new URI(dockerHost); - dataNodeHost = DockerClientConfigUtils.getDockerHostIpAddress(dockerHostUri); - - } catch (URISyntaxException e) { - LOG.warn("Using 'localhost' as the docker host because the URI '{}' for did not parse correctly: {}", - dockerHost, e.getMessage(), e); - } - } - } - - public DataNodeContainer(Network network) { - this(network, dataNodeHost); - } - - - - public DataNodeContainer(Network network, String name) { - super(network, name); - - withCommand("sh", "-c", "/hadoop/run-datanode.sh"); - - withExposedPorts(HDFSPorts.DATA_NODE_HTTP_PORT, HDFSPorts.DATA_NODE_DATA_TRANSFER_PORT, HDFSPorts.DATA_NODE_IPC_PORT); - - waitingFor(Wait.forHttp("/").forPort(HDFSPorts.DATA_NODE_HTTP_PORT)); - - - - /* - We need the name to be a valid hostname: the files are uploaded - directly to the dataNode host using the *hostname*. By default, the hostname - is not valid and no accessible from outside, therefore we trick the container - into using the localhost name so when the data node is resolved, it actually - points to the port on the local host that is redirected inside the container. - */ - withCreateContainerCmdModifier( - createContainerCmd -> { - createContainerCmd.withHostName(name); - createContainerCmd.withName(name); - } - ); - - addFixedExposedPort(HDFSPorts.DATA_NODE_HTTP_PORT, HDFSPorts.DATA_NODE_HTTP_PORT); - addFixedExposedPort(HDFSPorts.DATA_NODE_DATA_TRANSFER_PORT, HDFSPorts.DATA_NODE_DATA_TRANSFER_PORT); - addFixedExposedPort(HDFSPorts.DATA_NODE_IPC_PORT, HDFSPorts.DATA_NODE_IPC_PORT); - } - - public int getHttpPort() { - return getMappedPort(HDFSPorts.DATA_NODE_HTTP_PORT); - } - - public int getDataTransferPort() { - return HDFSPorts.DATA_NODE_DATA_TRANSFER_PORT; - } - - public int getIpcPort() { - return HDFSPorts.DATA_NODE_IPC_PORT; - } - - - -} diff --git a/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/services/HDFSPorts.java b/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/services/HDFSPorts.java deleted file mode 100644 index 190aef0..0000000 --- a/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/services/HDFSPorts.java +++ /dev/null @@ -1,31 +0,0 @@ -/* - * 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.camel.kafkaconnector.hdfs.services; - -final class HDFSPorts { - public static final int NAME_NODE_HTTP_PORT = 50070; - public static final int NAME_NODE_IPC_PORT = 8020; - - public static final int DATA_NODE_HTTP_PORT = 50075; - public static final int DATA_NODE_DATA_TRANSFER_PORT = 50010; - public static final int DATA_NODE_IPC_PORT = 50020; - - private HDFSPorts() { - } - -} diff --git a/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/services/HDFSService.java b/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/services/HDFSService.java deleted file mode 100644 index 0b78d98..0000000 --- a/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/services/HDFSService.java +++ /dev/null @@ -1,58 +0,0 @@ -/* - * 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.camel.kafkaconnector.hdfs.services; - -import org.junit.jupiter.api.extension.AfterAllCallback; -import org.junit.jupiter.api.extension.BeforeAllCallback; -import org.junit.jupiter.api.extension.ExtensionContext; - -public interface HDFSService extends BeforeAllCallback, AfterAllCallback { - - /** - * Gets the hostname of the HDFS server - * @return - */ - String getHDFSHost(); - - /** - * Gets the port used by the HDFS server - * @return - */ - int getPort(); - - /** - * Perform any initialization necessary - */ - void initialize(); - - /** - * Shuts down the service after the test has completed - */ - void shutdown(); - - - @Override - default void beforeAll(ExtensionContext extensionContext) throws Exception { - initialize(); - } - - @Override - default void afterAll(ExtensionContext extensionContext) throws Exception { - shutdown(); - } -} diff --git a/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/services/HDFSServiceFactory.java b/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/services/HDFSServiceFactory.java deleted file mode 100644 index 17e0319..0000000 --- a/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/services/HDFSServiceFactory.java +++ /dev/null @@ -1,46 +0,0 @@ -/* - * 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.camel.kafkaconnector.hdfs.services; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -public final class HDFSServiceFactory { - private static final Logger LOG = LoggerFactory.getLogger(HDFSServiceFactory.class); - - private HDFSServiceFactory() { - - } - - public static HDFSService createService() { - String instanceType = System.getProperty("hdfs.instance.type"); - - if (instanceType == null || instanceType.equals("local-hdfs-container")) { - return new ContainerLocalHDFSService(); - } - - if (instanceType.equals("remote")) { - return new RemoteHDFSService(); - } - - LOG.error("Invalid HDFS instance type: {}. Must be either 'remote' or 'local-hdfs-container", - instanceType); - throw new UnsupportedOperationException(String.format("Invalid HDFS instance type: %s", instanceType)); - - } -} diff --git a/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/services/HadoopBaseContainer.java b/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/services/HadoopBaseContainer.java deleted file mode 100644 index e01e478..0000000 --- a/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/services/HadoopBaseContainer.java +++ /dev/null @@ -1,48 +0,0 @@ -/* - * 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.camel.kafkaconnector.hdfs.services; - -import java.util.function.Consumer; - -import com.github.dockerjava.api.command.CreateContainerCmd; -import org.testcontainers.containers.GenericContainer; -import org.testcontainers.containers.Network; -import org.testcontainers.images.builder.ImageFromDockerfile; - -abstract class HadoopBaseContainer<T extends GenericContainer<T>> extends GenericContainer<T> { - - public HadoopBaseContainer(Network network, String name) { - super(new ImageFromDockerfile("hadoop-2x:ckc", false) - .withFileFromClasspath(".", - "org/apache/camel/kafkaconnector/hdfs/services/")); - - withNetwork(network); - - withCreateContainerCmdModifier( - new Consumer<CreateContainerCmd>() { - @Override - public void accept(CreateContainerCmd createContainerCmd) { - createContainerCmd.withHostName(name); - createContainerCmd.withName(name); - } - } - ); - } - - abstract int getHttpPort(); -} diff --git a/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/services/NameNodeContainer.java b/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/services/NameNodeContainer.java deleted file mode 100644 index 9b12e43..0000000 --- a/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/services/NameNodeContainer.java +++ /dev/null @@ -1,50 +0,0 @@ -/* - * 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.camel.kafkaconnector.hdfs.services; - -import org.testcontainers.containers.Network; -import org.testcontainers.containers.wait.strategy.Wait; - -public class NameNodeContainer extends HadoopBaseContainer<NameNodeContainer> { - - - - public NameNodeContainer(Network network) { - this(network, "namenode"); - } - - public NameNodeContainer(Network network, String name) { - super(network, name); - - withCommand("sh", "-c", "/hadoop/run-namenode.sh"); - withExposedPorts(HDFSPorts.NAME_NODE_HTTP_PORT, HDFSPorts.NAME_NODE_IPC_PORT); - - waitingFor(Wait.forHttp("/").forPort(HDFSPorts.NAME_NODE_HTTP_PORT)); - - addFixedExposedPort(HDFSPorts.NAME_NODE_HTTP_PORT, HDFSPorts.NAME_NODE_HTTP_PORT); - addFixedExposedPort(HDFSPorts.NAME_NODE_IPC_PORT, HDFSPorts.NAME_NODE_IPC_PORT); - } - - public int getHttpPort() { - return getMappedPort(HDFSPorts.NAME_NODE_HTTP_PORT); - } - - public int getIpcPort() { - return HDFSPorts.NAME_NODE_IPC_PORT; - } -} diff --git a/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/services/RemoteHDFSService.java b/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/services/RemoteHDFSService.java deleted file mode 100644 index 6717147..0000000 --- a/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/services/RemoteHDFSService.java +++ /dev/null @@ -1,47 +0,0 @@ -/* - * 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.camel.kafkaconnector.hdfs.services; - -public class RemoteHDFSService implements HDFSService { - - @Override - public void initialize() { - - } - - @Override - public void shutdown() { - - } - - @Override - public String getHDFSHost() { - return System.getProperty("hdfs.host"); - } - - @Override - public int getPort() { - String strPort = System.getProperty("hdfs.port"); - - if (strPort != null) { - return Integer.parseInt(strPort); - } - - return 8020; - } -} diff --git a/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/sink/CamelSinkHDFSITCase.java b/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/sink/CamelSinkHDFSITCase.java index e5a84b0..c102d15 100644 --- a/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/sink/CamelSinkHDFSITCase.java +++ b/tests/itests-hdfs/src/test/java/org/apache/camel/kafkaconnector/hdfs/sink/CamelSinkHDFSITCase.java @@ -24,9 +24,9 @@ import org.apache.camel.kafkaconnector.common.AbstractKafkaTest; import org.apache.camel.kafkaconnector.common.ConnectorPropertyFactory; import org.apache.camel.kafkaconnector.common.clients.kafka.KafkaClient; import org.apache.camel.kafkaconnector.common.utils.TestUtils; -import org.apache.camel.kafkaconnector.hdfs.services.HDFSService; -import org.apache.camel.kafkaconnector.hdfs.services.HDFSServiceFactory; import org.apache.camel.kafkaconnector.hdfs.utils.HDFSEasy; +import org.apache.camel.test.infra.hdfs.v2.services.HDFSService; +import org.apache.camel.test.infra.hdfs.v2.services.HDFSServiceFactory; import org.apache.hadoop.fs.LocatedFileStatus; import org.apache.hadoop.fs.Path; import org.junit.jupiter.api.AfterEach; @@ -36,14 +36,12 @@ import org.junit.jupiter.api.Timeout; import org.junit.jupiter.api.extension.RegisterExtension; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.testcontainers.junit.jupiter.Testcontainers; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; -@Testcontainers public class CamelSinkHDFSITCase extends AbstractKafkaTest { @RegisterExtension public static HDFSService hdfsService = HDFSServiceFactory.createService();