[SPARK-9065][Streaming][PySpark] Add MessageHandler for Kafka Python API

[zsxwing]

Fixed the merge conflicts in #7410

Closes #7410

[SparkQA]

Test build #46004 has finished for PR 9742 at commit 59df9ee.

• This patch fails PySpark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * case class PythonMessageAndMetadata(\n * class PythonMessageAndMetadataPickler extends IObjectPickler\n * class KafkaMessageAndMetadata(object):\n

[zsxwing]

@jerryshao After discussing with @tdas offline, I added some codes to speed up the case when the user doesn't provide a messageHandler. So that we can avoid message serialization for this case and won't reduce the performance. Could you take a look?

[jerryshao]

OK, let me check. But I'm wondering if there's a huge overhead for message serialization, since we only pack some of metadatas into message, don't do serialization-deserialization on the real message itself, not sure what's actual performance loss will happen.

[jerryshao]

LGTM.

[SparkQA]

Test build #46033 has finished for PR 9742 at commit 48aa7c3.

• This patch fails MiMa tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * case class PythonMessageAndMetadata(\n * class PythonMessageAndMetadataPickler extends IObjectPickler\n * class KafkaMessageAndMetadata(object):\n

[zsxwing]

retest this please

[tdas]

nit: incorrect indent.

[jerryshao]

Remove this default_message_handler

[SparkQA]

Test build #46045 has finished for PR 9742 at commit f7ad2e3.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #2068 has finished for PR 9742 at commit f7ad2e3.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * case class PythonMessageAndMetadata(\n * class PythonMessageAndMetadataPickler extends IObjectPickler\n * class KafkaMessageAndMetadata(object):\n

[SparkQA]

Test build #46040 has finished for PR 9742 at commit d3f135d.

• This patch fails from timeout after a configured wait of 250m.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * case class PythonMessageAndMetadata(\n * class PythonMessageAndMetadataPickler extends IObjectPickler\n * class KafkaMessageAndMetadata(object):\n

[zsxwing]

retest this please

[SparkQA]

Test build #46060 has finished for PR 9742 at commit f7ad2e3.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * case class PythonMessageAndMetadata(\n * class PythonMessageAndMetadataPickler extends IObjectPickler\n * class KafkaMessageAndMetadata(object):\n

[zsxwing]

retest this please

[SparkQA]

Test build #46097 has finished for PR 9742 at commit f7ad2e3.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * case class PythonMessageAndMetadata(\n * class PythonMessageAndMetadataPickler extends IObjectPickler\n * class KafkaMessageAndMetadata(object):\n * case class JSONOptions(\n

[SparkQA]

Test build #46115 has finished for PR 9742 at commit b2bbfee.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * public class JavaLinearRegressionWithElasticNetExample\n * public class JavaLogisticRegressionSummaryExample\n * public class JavaLogisticRegressionWithElasticNetExample\n * case class PythonMessageAndMetadata(\n * class PythonMessageAndMetadataPickler extends IObjectPickler\n * class HashingTF(override val uid: String)\n * class Interaction @Since(\"1.6.0\") (override val uid: String) extends Transformer\n * class Normalizer(override val uid: String)\n * class SQLTransformer @Since(\"1.6.0\") (override val uid: String) extends Transformer with Writable\n * class Tokenizer(override val uid: String)\n * class KafkaMessageAndMetadata(object):\n * case class ArrayConversion(elementConversion: JDBCConversion) extends JDBCConversion\n * abstract class JdbcDialect extends Serializable\n

[tdas]

All right. I am merging to master and 1.6. Thanks @jerryshao for all the effort, and @zsxwing for getting this ready.

[jerryshao]

Thanks a lot @tdas and @zsxwing to make it merged, appreciated.