[
https://issues.apache.org/jira/browse/DRILL-4779?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16256345#comment-16256345
]
ASF GitHub Bot commented on DRILL-4779:
---------------------------------------
Github user paul-rogers commented on a diff in the pull request:
https://github.com/apache/drill/pull/1027#discussion_r151586477
--- Diff:
contrib/storage-kafka/src/main/java/org/apache/drill/exec/store/kafka/decoders/MessageReaderFactory.java
---
@@ -0,0 +1,56 @@
+/*
+ * 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.drill.exec.store.kafka.decoders;
+
+import org.apache.drill.common.exceptions.UserException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Preconditions;
+
+public class MessageReaderFactory {
+
+ private static final Logger logger =
LoggerFactory.getLogger(MessageReaderFactory.class);
+
+ /**
+ * Initialize kafka message reader beased on store.kafka.record.reader
session
+ * property
+ *
+ * @param messageReaderKlass
+ * value of store.kafka.record.reader session property
+ * @return kafka message reader
+ * @throws UserException
+ * in case of any message reader initialization
+ */
+ public static MessageReader getMessageReader(String messageReaderKlass) {
+ Preconditions.checkNotNull(messageReaderKlass, "Please set
store.kafka.record.reader " + messageReaderKlass);
+ MessageReader messageReader = null;
+ try {
+ Class<?> klass = Class.forName(messageReaderKlass);
--- End diff --
Actually, seeing the larger picture, I wonder if this is the right
approach. Suppose this is two or three releases from now and we support other
forms of Kafka messages. Different topics use different formats.
If the message format is a system/session option, then I need to switch the
option before each query. Very cumbersome and error prone.
Instead, perhaps this information should be part of the storage plugin
config. Then, I can define different plugins: one for each message format.
Further, can I have multiple Kafka servers? If so, would I need different
plugin configs for each?
So, should we be thinking about encoding most properties as plugin config
properties?
Now, the plugin might have a `format` property, one of which is `json`. The
JSON config properties would be defined in the `json` format within the overall
storage plugin config.
Given the looming deadline of the 1.12 release, it is fine if we file the
above suggestion as a JIRA and work on it for 1.13.
> Kafka storage plugin support
> ----------------------------
>
> Key: DRILL-4779
> URL: https://issues.apache.org/jira/browse/DRILL-4779
> Project: Apache Drill
> Issue Type: New Feature
> Components: Storage - Other
> Affects Versions: 1.11.0
> Reporter: B Anil Kumar
> Assignee: B Anil Kumar
> Labels: doc-impacting
> Fix For: 1.12.0
>
>
> Implement Kafka storage plugin will enable the strong SQL support for Kafka.
> Initially implementation can target for supporting json and avro message types
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
