@@ -0,0 +1,2377 @@
+/*%0A * Licensed to the Apache Software Foundation (ASF) under one or more%0A * contributor license agreements.  See the NOTICE file distributed with%0A * this work for additional information regarding copyright ownership.%0A * The ASF licenses this file to You under the Apache License, Version 2.0%0A * (the %22License%22); you may not use this file except in compliance with%0A * the License.  You may obtain a copy of the License at%0A *%0A *     http://www.apache.org/licenses/LICENSE-2.0%0A *%0A *  Unless required by applicable law or agreed to in writing, software%0A *  distributed under the License is distributed on an %22AS IS%22 BASIS,%0A *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.%0A *  See the License for the specific language governing permissions and%0A *  limitations under the License.%0A *%0A */%0A%0Apackage org.apache.eagle.security.auditlog.kafka;%0A%0Aimport backtype.storm.spout.Scheme;%0Aimport backtype.storm.tuple.Fields;%0Aimport backtype.storm.tuple.Values;%0Aimport org.codehaus.jackson.map.ObjectMapper;%0Aimport org.slf4j.Logger;%0Aimport storm.kafka.StringScheme;%0A%0Aimport java.io.IOException;%0Aimport java.util.List;%0Aimport java.util.Map;%0A%0Apublic class MessageJsonScheme  implements Scheme %7B%0A%0A    private static final Logger LOG = org.slf4j.LoggerFactory.getLogger(MessageJsonScheme.class);%0A    private static final ObjectMapper mapper = new ObjectMapper();%0A%0A    public static final String MESSAGE_SCHEME_KEY = %22message%22;%0A%0A    @Override%0A    public Fields getOutputFields() %7B%0A        return new Fields(StringScheme.STRING_SCHEME_KEY);%0A    %7D%0A%0A    @Override%0A    @SuppressWarnings(%22rawtypes%22)%0A    public List%3CObject%3E deserialize(byte%5B%5D ser) %7B%0A        try %7B%0A            if (ser != null) %7B%0A                Map map = mapper.readValue(ser, Map.class);%0A                Object message = map.get(MESSAGE_SCHEME_KEY);%0A                if (message != null) %7B%0A                    return new Values(map.get(MESSAGE_SCHEME_KEY));%0A                %7D%0A            %7D else %7B%0A                if (LOG.isDebugEnabled()) %7B%0A                    LOG.debug(%22Content is null, ignore%22);%0A                %7D%0A            %7D%0A        %7D catch (IOException e) %7B%0A            try %7B%0A                LOG.error(%22Failed to deserialize as JSON: %7B%7D%22, new String(ser, %22UTF-8%22), e);%0A            %7D catch (Exception ex) %7B%0A                LOG.error(ex.getMessage(), ex);%0A            %7D%0A        %7D%0A        return null;%0A    %7D%0A%7D

add MessageJsonScheme which extracts audit log from a json formatted …