Merge pull request #175 from cloudsufi/bump/cdap-spark

Bump cdap and hadoop version
data-integrations · Dec 5, 2024 · 5a3f8ab · 5a3f8ab
2 parents 008abe4 + 7f7e8c1
commit 5a3f8ab
Show file tree

Hide file tree

Showing 5 changed files with 22 additions and 25 deletions.
diff --git a/pom.xml b/pom.xml
@@ -76,20 +76,20 @@
 
   <properties>
     <awaitility.version>3.1.6</awaitility.version>
-    <cdap.version>6.8.0-SNAPSHOT</cdap.version>
+    <cdap.version>6.11.0-SNAPSHOT</cdap.version>
     <commons.version>3.9</commons.version>
     <common.codec.version>1.12</common.codec.version>
     <gson.version>2.8.5</gson.version>
     <googleauth.version>0.4.0</googleauth.version>
-    <hadoop.version>2.3.0</hadoop.version>
+    <hadoop.version>3.3.6</hadoop.version>
     <httpcomponents.version>4.5.9</httpcomponents.version>
-    <hydrator.version>2.10.0-SNAPSHOT</hydrator.version>
+    <hydrator.version>2.13.0-SNAPSHOT</hydrator.version>
     <cdap.plugin.version>2.12.0</cdap.plugin.version>
-    <jackson.version>2.9.9</jackson.version>
+    <jackson.version>2.13.0</jackson.version>
     <junit.version>4.11</junit.version>
     <jython.version>2.7.1</jython.version>
     <mockito.version>2.24.0</mockito.version>
-    <spark2.version>2.1.3</spark2.version>
+    <spark3.version>3.3.2</spark3.version>
     <unxml.version>0.9</unxml.version>
     <wiremock.version>1.49</wiremock.version>
   </properties>
@@ -150,6 +150,10 @@
           <groupId>org.slf4j</groupId>
           <artifactId>slf4j-log4j12</artifactId>
         </exclusion>
+        <exclusion>
+          <groupId>org.slf4j</groupId>
+          <artifactId>slf4j-reload4j</artifactId>
+        </exclusion>
         <exclusion>
           <groupId>org.apache.avro</groupId>
           <artifactId>avro</artifactId>
@@ -212,7 +216,7 @@
       <exclusions>
         <exclusion>
           <groupId>org.slf4j</groupId>
-          <artifactId>slf4j-log4j12</artifactId>
+          <artifactId>slf4j-reload4j</artifactId>
         </exclusion>
         <exclusion>
           <groupId>com.google.inject.extensions</groupId>
@@ -253,14 +257,14 @@
     </dependency>
     <dependency>
       <groupId>org.apache.spark</groupId>
-      <artifactId>spark-streaming_2.11</artifactId>
-      <version>${spark2.version}</version>
+      <artifactId>spark-streaming_2.12</artifactId>
+      <version>${spark3.version}</version>
       <scope>provided</scope>
     </dependency>
     <dependency>
       <groupId>org.apache.spark</groupId>
-      <artifactId>spark-core_2.11</artifactId>
-      <version>${spark2.version}</version>
+      <artifactId>spark-core_2.12</artifactId>
+      <version>${spark3.version}</version>
       <scope>provided</scope>
       <exclusions>
         <exclusion>
@@ -410,19 +414,19 @@
     </dependency>
     <dependency>
       <groupId>io.cdap.cdap</groupId>
-      <artifactId>cdap-data-streams2_2.11</artifactId>
+      <artifactId>cdap-data-streams3_2.12</artifactId>
       <version>${cdap.version}</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>io.cdap.cdap</groupId>
-      <artifactId>cdap-data-pipeline2_2.11</artifactId>
+      <artifactId>cdap-data-pipeline3_2.12</artifactId>
       <version>${cdap.version}</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>io.cdap.cdap</groupId>
-      <artifactId>cdap-spark-core2_2.11</artifactId>
+      <artifactId>cdap-spark-core3_2.12</artifactId>
       <version>${cdap.version}</version>
       <scope>test</scope>
     </dependency>

diff --git a/src/main/java/io/cdap/plugin/http/sink/batch/HTTPSink.java b/src/main/java/io/cdap/plugin/http/sink/batch/HTTPSink.java
@@ -102,8 +102,10 @@ public String getOutputFormatClassName() {
 
     @Override
     public Map<String, String> getOutputFormatConfiguration() {
+      Schema defaultValidSchema = Schema.recordOf("schema", Schema.Field.of("body", Schema.of(Schema.Type.STRING)));
       return ImmutableMap.of("http.sink.config", GSON.toJson(config),
-                             "http.sink.input.schema", inputSchema == null ? "" : inputSchema.toString());
+                             "http.sink.input.schema",
+                             inputSchema == null ? defaultValidSchema.toString() : inputSchema.toString());
     }
   }
 

diff --git a/src/test/java/io/cdap/plugin/http/etl/HttpSourceETLTest.java b/src/test/java/io/cdap/plugin/http/etl/HttpSourceETLTest.java
@@ -135,6 +135,7 @@ public void testIncrementAnIndexXml() throws Exception {
       .put(BaseHttpSourceConfig.PROPERTY_PAGINATION_TYPE, "Increment an index")
       .put(BaseHttpSourceConfig.PROPERTY_START_INDEX, "0")
       .put(BaseHttpSourceConfig.PROPERTY_INDEX_INCREMENT, "20")
+      .put(BaseHttpSourceConfig.PROPERTY_MAX_INDEX, "100")
       .build();
 
     wireMockRule.stubFor(WireMock.get(

diff --git a/src/test/java/io/cdap/plugin/http/etl/HttpStreamingSourceETLTest.java b/src/test/java/io/cdap/plugin/http/etl/HttpStreamingSourceETLTest.java
@@ -19,15 +19,13 @@
 import io.cdap.cdap.api.artifact.ArtifactSummary;
 import io.cdap.cdap.api.data.format.StructuredRecord;
 import io.cdap.cdap.api.dataset.table.Table;
-import io.cdap.cdap.common.conf.Constants;
 import io.cdap.cdap.datastreams.DataStreamsApp;
 import io.cdap.cdap.datastreams.DataStreamsSparkLauncher;
 import io.cdap.cdap.etl.api.streaming.StreamingSource;
 import io.cdap.cdap.etl.mock.batch.MockSink;
 import io.cdap.cdap.etl.proto.v2.DataStreamsConfig;
 import io.cdap.cdap.etl.proto.v2.ETLPlugin;
 import io.cdap.cdap.etl.proto.v2.ETLStage;
-import io.cdap.cdap.etl.spark.Compat;
 import io.cdap.cdap.proto.ProgramRunStatus;
 import io.cdap.cdap.proto.artifact.AppRequest;
 import io.cdap.cdap.proto.id.ApplicationId;
@@ -37,12 +35,10 @@
 import io.cdap.cdap.test.DataSetManager;
 import io.cdap.cdap.test.ProgramManager;
 import io.cdap.cdap.test.SparkManager;
-import io.cdap.cdap.test.TestConfiguration;
 import io.cdap.plugin.http.source.streaming.HttpStreamingSource;
 import org.awaitility.Awaitility;
 import org.junit.Assert;
 import org.junit.BeforeClass;
-import org.junit.ClassRule;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -57,11 +53,6 @@ public class HttpStreamingSourceETLTest extends HttpSourceETLTest {
   private static final int WAIT_FOR_RECORDS_TIMEOUT_SECONDS = 60;
   private static final long WAIT_FOR_RECORDS_POLLING_INTERVAL_MS = 100;
 
-  @ClassRule
-  public static final TestConfiguration CONFIG =
-    new TestConfiguration(Constants.Explore.EXPLORE_ENABLED, false,
-                          Constants.AppFabric.SPARK_COMPAT, Compat.SPARK_COMPAT);
-
   @BeforeClass
   public static void setupTest() throws Exception {
     LOG.info("Setting up application");

diff --git a/src/test/java/io/cdap/plugin/http/sink/batch/HTTPSinkTest.java b/src/test/java/io/cdap/plugin/http/sink/batch/HTTPSinkTest.java
@@ -75,8 +75,7 @@ public class HTTPSinkTest extends HydratorTestBase {
   protected static final ArtifactId BATCH_ARTIFACT_ID = NamespaceId.DEFAULT.artifact("data-pipeline", "4.0.0");
   protected static final ArtifactSummary BATCH_ARTIFACT = new ArtifactSummary("data-pipeline", "4.0.0");
   private static final Schema inputSchema = Schema.recordOf(
-    "input-record",
-    Schema.Field.of("id", Schema.of(Schema.Type.STRING)));
+    "input-record", Schema.Field.of("id", Schema.of(Schema.Type.STRING)));
   private static NettyHttpService httpService;
   protected static String baseURL;