Merge pull request #19 from Alvearie/release-3.2.0

Release 3.2.0

Author: alisakotliarova

.github/workflows/ci-workflow.yml

@@ -28,7 +28,7 @@ jobs:
         uses: actions/setup-java@v2
         with:
           java-version: '8'
-          distribution: 'adopt'
+          distribution: 'temurin'
           cache: 'gradle'
 
       - name: Set GitHub Tag environment variable

.github/workflows/code-scans.yml

@@ -22,11 +22,11 @@ jobs:
           # full checkout for SonarCloud analysis
           fetch-depth: 0
 
-      - name: Set up Java 11
+      - name: Set up Java jdk11
         uses: actions/setup-java@v2
         with:
           java-version: '11'
-          distribution: 'adopt'
+          distribution: 'temurin'
           cache: 'gradle'
 
       - name: SonarCloud Scan

.github/workflows/pull_request.yml

@@ -15,8 +15,6 @@ jobs:
     # run if the PR was opened or there was a comment containing '/pr_checks'
     if: (github.event_name == 'pull_request') || (github.event_name == 'pull_request_review' && contains(github.event.review.body, '/pr_checks'))
     runs-on: ubuntu-latest
-    env:
-      RESOURCE_GROUP: hri-dev1-wdc-kube
 
     steps:
       - name: Set branch name environment variable
@@ -28,7 +26,7 @@ jobs:
         uses: actions/setup-java@v2
         with:
           java-version: '8'
-          distribution: 'adopt-hotspot'
+          distribution: 'temurin'
           cache: 'gradle'
 
       - name: Gradle build

build.gradle

@@ -24,12 +24,12 @@ plugins {
 }
 
 group = 'org.alvearie.hri.flink'
-version = '3.1-1.0.3' // Update version number in releases
+version = '3.2-1.1.0' // Update version number in releases
 description = """HRI Flink Pipeline Core Library"""
 
 ext {
     javaVersion = '1.8'
-    flinkVersion = '1.10.3'
+    flinkVersion = '1.14.0'
     scalaBinaryVersion = '2.12'
     scalaVersion = '2.12.11'
     scalaTestVersion = '3.1.1'
@@ -121,12 +121,6 @@ ossIndexAudit {
             '32bccbbb-bd3f-44f3-a530-5827a93d74db', '2418a058-dfb7-4303-b1d1-ac5262d9e499', '1be5c4d8-4994-4dd6-92b1-ca53ac7e6a17', 'f984e75c-06bf-43dc-86db-3f8dfe82f430',
             'bd0f0dd9-1356-4dec-a8a3-edc777cc92d5', 'b70e447a-06a1-48b0-a02b-cd5b78862777',
 
-            //pkg:maven/org.apache.commons/commons-compress@1.18 (from org.apache.flink:flink-scala_2.12:1.10.3)
-            //NOTE: these vulnerabilities cannot be eliminated UNTIL we upgrade our Flink version to Flink 1.14.0
-            //Need to get Security Exception until that happens
-            '68232267-bb25-4b04-8dec-caf7c11c7293', '69b8043a-3002-48fa-9762-8f6040d83de1', '4102317d-8250-465e-a46d-179d42792b14',
-            '7a6a9dd2-67de-4e2a-b406-7aa4a4ce29cc', '8ea14e38-e6cc-48d9-bfe4-ec89f93596e7',
-
             //pkg:maven/org.apache.logging.log4j/log4j-core@2.11.2 (from org.scala-sbt:zinc_2.12:1.3.5 - Used by gradle scala plugin only)
             'd3477f9c-032a-44a7-a5e1-02ae35e4737c',
 
@@ -178,7 +172,7 @@ dependencies {
     // Dependencies that library users should include in their job
     // shadow jar
     // --------------------------------------------------------------
-    implementation "org.alvearie.hri:hri-api-batch-notification:3.1-2.0.2"
+    implementation "org.alvearie.hri:hri-api-batch-notification:3.2-2.0.3"
     implementation "org.apache.flink:flink-connector-kafka_${scalaBinaryVersion}:${flinkVersion}"
     implementation "com.fasterxml.jackson.core:jackson-databind:${jacksonVersion}"
     implementation "com.fasterxml.jackson.module:jackson-module-scala_${scalaBinaryVersion}:${jacksonVersion}"
@@ -193,7 +187,7 @@ dependencies {
     testImplementation "org.mockito:mockito-scala-scalatest_${scalaBinaryVersion}:1.14.4"
     testImplementation "org.apache.flink:flink-tests:${flinkVersion}:tests"
     testImplementation "org.apache.flink:flink-test-utils_${scalaBinaryVersion}:${flinkVersion}"
-    testImplementation "org.apache.flink:flink-runtime_${scalaBinaryVersion}:${flinkVersion}:tests"
+    testImplementation "org.apache.flink:flink-runtime:${flinkVersion}:tests"
     testImplementation "org.apache.flink:flink-streaming-java_${scalaBinaryVersion}:${flinkVersion}:tests"
 
 

src/main/scala/org/alvearie/hri/flink/core/MgmtApiSink.scala

@@ -40,7 +40,7 @@ class MgmtApiSink(val tenantId: String, val mgmtApiUrl: String, val mgmtClientId
     }
   }
 
-  override def invoke(record: NotificationRecord, context: Context[_]): Unit = {
+  override def invoke(record: NotificationRecord, context: Context): Unit = {
     val batch = record.value
     log.info("Received Batch {} with status of {}", batch.getId: Any, batch.getStatus: Any)
     batch.getStatus match {

src/test/scala/org/alvearie/hri/flink/core/serialization/HriRecordDeserializerTest.scala

@@ -60,7 +60,7 @@ class HriRecordDeserializerTest extends AnyFunSuite {
 
     test("Null headers should be preserved") {
         val deserializer = new HriRecordDeserializer
-        val consumerRecord = createConsumerRecord(null, FakeKey, FakeBody)
+        val consumerRecord = createConsumerRecordWithoutHeaders(FakeKey, FakeBody)
 
         val record = deserializer.deserialize(consumerRecord)
         val actualHeaders = record.headers
@@ -70,7 +70,7 @@ class HriRecordDeserializerTest extends AnyFunSuite {
         val actualPartition = record.partition
         val actualOffset = record.offset
 
-        actualHeaders should be(null)
+        actualHeaders.toArray shouldBe empty
         actualKey should equal(ExpectedKey)
         actualBody should equal(ExpectedBody)
         actualTopic should equal(TopicName)
@@ -80,7 +80,7 @@ class HriRecordDeserializerTest extends AnyFunSuite {
 
     test("Null key should be preserved") {
         val deserializer = new HriRecordDeserializer
-        val consumerRecord = createConsumerRecord(null, null, FakeBody)
+        val consumerRecord = createConsumerRecordWithoutHeaders(null, FakeBody)
 
         val record = deserializer.deserialize(consumerRecord)
         val actualHeaders = record.headers
@@ -90,7 +90,7 @@ class HriRecordDeserializerTest extends AnyFunSuite {
         val actualPartition = record.partition
         val actualOffset = record.offset
 
-        actualHeaders should be(null)
+        actualHeaders.toArray shouldBe empty
         actualKey should be(null)
         actualBody should equal(ExpectedBody)
         actualTopic should equal(TopicName)
@@ -100,7 +100,7 @@ class HriRecordDeserializerTest extends AnyFunSuite {
 
     test("Null body should be preserved") {
         val deserializer = new HriRecordDeserializer
-        val consumerRecord = createConsumerRecord(null, null, null)
+        val consumerRecord = createConsumerRecordWithoutHeaders(null, null)
 
         val record = deserializer.deserialize(consumerRecord)
         val actualHeaders = record.headers
@@ -110,7 +110,7 @@ class HriRecordDeserializerTest extends AnyFunSuite {
         val actualPartition = record.partition
         val actualOffset = record.offset
 
-        actualHeaders should be(null)
+        actualHeaders.toArray shouldBe empty
         actualKey should be(null)
         actualBody should be(null)
         actualTopic should equal(TopicName)
@@ -170,4 +170,18 @@ class HriRecordDeserializerTest extends AnyFunSuite {
         )
     }
 
+    def createConsumerRecordWithoutHeaders(key: Array[Byte], value: Array[Byte]): ConsumerRecord[Array[Byte], Array[Byte]] = {
+        new ConsumerRecord[Array[Byte], Array[Byte]](
+            TopicName,
+            Partition,
+            Offset,
+            ConsumerRecord.NO_TIMESTAMP,
+            TimestampType.NO_TIMESTAMP_TYPE,
+            ConsumerRecord.NULL_CHECKSUM.toLong,
+            ConsumerRecord.NULL_SIZE,
+            ConsumerRecord.NULL_SIZE,
+            key,
+            value,
+        )
+    }
 }

src/test/scala/org/alvearie/hri/flink/core/serialization/NotificationDeserializerTest.scala

@@ -67,42 +67,42 @@ class NotificationDeserializerTest extends AnyFunSuite {
 
     test("Null headers should be preserved") {
         val deserializer = new NotificationDeserializer()
-        val consumerRecord = createConsumerRecord(null, FakeKey, FakeNotification)
+        val consumerRecord = createConsumerRecordWithoutHeaders(FakeKey, FakeNotification)
 
         val record = deserializer.deserialize(consumerRecord)
         val actualHeaders = record.headers
         val actualKey = new String(record.key, StandardCharsets.UTF_8)
         val actualNotification = record.value
 
-        actualHeaders should be(null)
+        actualHeaders.toArray shouldBe empty
         actualKey should equal(ExpectedKey)
         actualNotification should equal(ExpectedNotification)
     }
 
     test("Null key should be preserved") {
         val deserializer = new NotificationDeserializer()
-        val consumerRecord = createConsumerRecord(null, null, FakeNotification)
+        val consumerRecord = createConsumerRecordWithoutHeaders(null, FakeNotification)
 
         val record = deserializer.deserialize(consumerRecord)
         val actualHeaders = record.headers
         val actualKey = record.key
         val actualNotification = record.value
 
-        actualHeaders should be(null)
+        actualHeaders.toArray shouldBe empty
         actualKey should be(null)
         actualNotification should equal(ExpectedNotification)
     }
 
     test("Null body should be preserved") {
         val deserializer = new NotificationDeserializer()
-        val consumerRecord = createConsumerRecord(null, null, null)
+        val consumerRecord = createConsumerRecordWithoutHeaders(null, null)
 
         val record = deserializer.deserialize(consumerRecord)
         val actualHeaders = record.headers
         val actualKey = record.key
         val actualNotification = record.value
 
-        actualHeaders should be(null)
+        actualHeaders.toArray shouldBe empty
         actualKey should be(null)
         actualNotification should equal(null)
     }
@@ -138,7 +138,6 @@ class NotificationDeserializerTest extends AnyFunSuite {
         headers
     }
 
-
     def createConsumerRecord(headers: Headers, key: Array[Byte], value: Array[Byte]): ConsumerRecord[Array[Byte], Array[Byte]] = {
         new ConsumerRecord[Array[Byte], Array[Byte]](
             TopicName,
@@ -155,6 +154,21 @@ class NotificationDeserializerTest extends AnyFunSuite {
         )
     }
 
+    def createConsumerRecordWithoutHeaders(key: Array[Byte], value: Array[Byte]): ConsumerRecord[Array[Byte], Array[Byte]] = {
+        new ConsumerRecord[Array[Byte], Array[Byte]](
+            TopicName,
+            Partition,
+            Offset,
+            ConsumerRecord.NO_TIMESTAMP,
+            TimestampType.NO_TIMESTAMP_TYPE,
+            ConsumerRecord.NULL_CHECKSUM.toLong,
+            ConsumerRecord.NULL_SIZE,
+            ConsumerRecord.NULL_SIZE,
+            key,
+            value,
+        )
+    }
+
     def createNotification(): BatchNotification = {
          new BatchNotification()
           .withId("id")
@@ -170,5 +184,4 @@ class NotificationDeserializerTest extends AnyFunSuite {
           .withInvalidThreshold(10)
           .withFailureMessage("")
     }
-
 }