Skip to content
Snippets Groups Projects
Commit 82138e09 authored by Burak Yavuz's avatar Burak Yavuz Committed by Shixiong Zhu
Browse files

[SPARK-19886] Fix reportDataLoss if statement in SS KafkaSource

## What changes were proposed in this pull request?

Fix the `throw new IllegalStateException` if statement part.

## How is this patch tested

Regression test

Author: Burak Yavuz <brkyvz@gmail.com>

Closes #17228 from brkyvz/kafka-cause-fix.
parent f79371ad
No related branches found
No related tags found
No related merge requests found
......@@ -273,19 +273,7 @@ private[kafka010] case class CachedKafkaConsumer private(
message: String,
cause: Throwable = null): Unit = {
val finalMessage = s"$message ${additionalMessage(failOnDataLoss)}"
if (failOnDataLoss) {
if (cause != null) {
throw new IllegalStateException(finalMessage)
} else {
throw new IllegalStateException(finalMessage, cause)
}
} else {
if (cause != null) {
logWarning(finalMessage)
} else {
logWarning(finalMessage, cause)
}
}
reportDataLoss0(failOnDataLoss, finalMessage, cause)
}
private def close(): Unit = consumer.close()
......@@ -398,4 +386,23 @@ private[kafka010] object CachedKafkaConsumer extends Logging {
consumer
}
}
private def reportDataLoss0(
failOnDataLoss: Boolean,
finalMessage: String,
cause: Throwable = null): Unit = {
if (failOnDataLoss) {
if (cause != null) {
throw new IllegalStateException(finalMessage, cause)
} else {
throw new IllegalStateException(finalMessage)
}
} else {
if (cause != null) {
logWarning(finalMessage, cause)
} else {
logWarning(finalMessage)
}
}
}
}
/*
* 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.spark.sql.kafka010
import org.scalatest.PrivateMethodTester
import org.apache.spark.sql.test.SharedSQLContext
class CachedKafkaConsumerSuite extends SharedSQLContext with PrivateMethodTester {
test("SPARK-19886: Report error cause correctly in reportDataLoss") {
val cause = new Exception("D'oh!")
val reportDataLoss = PrivateMethod[Unit]('reportDataLoss0)
val e = intercept[IllegalStateException] {
CachedKafkaConsumer.invokePrivate(reportDataLoss(true, "message", cause))
}
assert(e.getCause === cause)
}
}
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment