We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
handle throttles on log group output


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `c7n/log.py`
Content:
```
1 # Copyright 2016 Capital One Services, LLC
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 # http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 """
15 Python Standard Logging integration with CloudWatch Logs
16 
17 Double Buffered with background thread delivery.
18 
19 We do an initial buffering on the log handler directly, to avoid
20 some of the overhead of pushing to the queue (albeit dubious as
21 std logging does default lock acquisition around handler emit).
22 also uses a single thread for all outbound. Background thread
23 uses a separate session.
24 """
25 
26 import boto3
27 from botocore.exceptions import ClientError
28 
29 import itertools
30 import logging
31 from operator import itemgetter
32 import threading
33 import time
34 
35 try:
36     import Queue
37 except ImportError:
38     import queue as Queue
39 
40 
41 FLUSH_MARKER = object()
42 SHUTDOWN_MARKER = object()
43 
44 EMPTY = Queue.Empty
45 
46 
47 class Error(object):
48 
49     AlreadyAccepted = "DataAlreadyAcceptedException"
50     InvalidToken = "InvalidSequenceTokenException"
51     ResourceExists = "ResourceAlreadyExistsException"
52 
53     @staticmethod
54     def code(e):
55         return e.response.get('Error', {}).get('Code')
56 
57 
58 class CloudWatchLogHandler(logging.Handler):
59     """Python Log Handler to Send to Cloud Watch Logs
60 
61     http://goo.gl/eZGAEK
62     """
63 
64     batch_size = 20
65     batch_interval = 40
66     batch_min_buffer = 10
67 
68     def __init__(self, log_group=__name__, log_stream=None,
69                  session_factory=None):
70         super(CloudWatchLogHandler, self).__init__()
71         self.log_group = log_group
72         self.log_stream = log_stream
73         self.session_factory = session_factory or boto3.Session
74         self.transport = None
75         self.queue = Queue.Queue()
76         self.threads = []
77         # do some basic buffering before sending to transport to minimize
78         # queue/threading overhead
79         self.buf = []
80         self.last_seen = time.time()
81         # Logging module internally is tracking all handlers, for final
82         # cleanup atexit, custodian is a bit more explicitly scoping shutdown to
83         # each policy, so use a sentinel value to avoid deadlocks.
84         self.shutdown = False
85         try:
86             self.session_factory().client(
87                 'logs').create_log_group(logGroupName=self.log_group)
88         except ClientError as e:
89             if Error.code(e) != Error.ResourceExists:
90                 raise
91 
92     # Begin logging.Handler API
93     def emit(self, message):
94         """Send logs"""
95         # We're sending messages asynchronously, bubble to caller when
96         # we've detected an error on the message. This isn't great,
97         # but options once we've gone async without a deferred/promise
98         # aren't great.
99         if self.transport and self.transport.error:
100             raise self.transport.error
101 
102         # Sanity safety, people do like to recurse by attaching to
103         # root log :-(
104         if message.name.startswith('boto'):
105             return
106 
107         msg = self.format_message(message)
108         if not self.transport:
109             self.start_transports()
110         self.buf.append(msg)
111         self.flush_buffers(
112             (message.created - self.last_seen >= self.batch_interval))
113 
114         self.last_seen = message.created
115 
116     def flush(self):
117         """Ensure all logging output has been flushed."""
118         if self.shutdown:
119             return
120         self.flush_buffers(force=True)
121         self.queue.put(FLUSH_MARKER)
122         self.queue.join()
123 
124     def close(self):
125         if self.shutdown:
126             return
127         self.shutdown = True
128         self.queue.put(SHUTDOWN_MARKER)
129         self.queue.join()
130         for t in self.threads:
131             t.join()
132         self.threads = []
133 
134     # End logging.Handler API
135 
136     def format_message(self, msg):
137         """format message."""
138         return {'timestamp': int(msg.created * 1000),
139                 'message': self.format(msg),
140                 'stream': self.log_stream or msg.name,
141                 'group': self.log_group}
142 
143     def start_transports(self):
144         """start thread transports."""
145         self.transport = Transport(
146             self.queue, self.batch_size, self.batch_interval,
147             self.session_factory)
148         thread = threading.Thread(target=self.transport.loop)
149         self.threads.append(thread)
150         thread.daemon = True
151         thread.start()
152 
153     def flush_buffers(self, force=False):
154         if not force and len(self.buf) < self.batch_min_buffer:
155             return
156         self.queue.put(self.buf)
157         self.buf = []
158 
159 
160 class Transport(object):
161 
162     def __init__(self, queue, batch_size, batch_interval, session_factory):
163         self.queue = queue
164         self.batch_size = batch_size
165         self.batch_interval = batch_interval
166         self.client = session_factory().client('logs')
167         self.sequences = {}
168         self.buffers = {}
169         self.error = None
170 
171     def create_stream(self, group, stream):
172         try:
173             self.client.create_log_stream(
174                 logGroupName=group, logStreamName=stream)
175         except ClientError as e:
176             if Error.code(e) != Error.ResourceExists:
177                 self.error = e
178                 return False
179         return True
180 
181     def send(self):
182         for k, messages in self.buffers.items():
183             self.send_group(k, messages)
184         self.buffers = {}
185 
186     def send_group(self, k, messages):
187         group, stream = k.split('=', 1)
188         if stream not in self.sequences:
189             if not self.create_stream(group, stream):
190                 return
191             self.sequences[stream] = None
192         params = dict(
193             logGroupName=group, logStreamName=stream,
194             logEvents=sorted(
195                 messages, key=itemgetter('timestamp'), reverse=False))
196         if self.sequences[stream]:
197             params['sequenceToken'] = self.sequences[stream]
198         try:
199             response = self.client.put_log_events(**params)
200         except ClientError as e:
201             if Error.code(e) in (Error.AlreadyAccepted, Error.InvalidToken):
202                 self.sequences[stream] = e.response['Error']['Message'].rsplit(
203                     " ", 1)[-1]
204                 return self.send_group(k, messages)
205             self.error = e
206             return
207         self.sequences[stream] = response['nextSequenceToken']
208 
209     def loop(self):
210         def keyed(datum):
211             return "%s=%s" % (
212                 datum.pop('group'), datum.pop('stream'))
213 
214         while True:
215             try:
216                 datum = self.queue.get(block=True, timeout=self.batch_interval)
217             except EMPTY:
218                 if Queue is None:
219                     return
220                 datum = None
221             if datum is None:
222                 # Timeout reached, flush
223                 self.send()
224                 continue
225             elif datum == FLUSH_MARKER:
226                 self.send()
227             elif datum == SHUTDOWN_MARKER:
228                 self.queue.task_done()
229                 return
230             else:
231                 for k, group in itertools.groupby(datum, keyed):
232                     self.buffers.setdefault(k, []).extend(group)
233             self.queue.task_done()
234
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/c7n/log.py b/c7n/log.py
--- a/c7n/log.py
+++ b/c7n/log.py
@@ -37,6 +37,7 @@
 except ImportError:
     import queue as Queue
 
+from c7n.utils import get_retry
 
 FLUSH_MARKER = object()
 SHUTDOWN_MARKER = object()
@@ -82,9 +83,15 @@
         # cleanup atexit, custodian is a bit more explicitly scoping shutdown to
         # each policy, so use a sentinel value to avoid deadlocks.
         self.shutdown = False
+        retry = get_retry(('ThrottlingException',))
         try:
-            self.session_factory().client(
-                'logs').create_log_group(logGroupName=self.log_group)
+            client = self.session_factory().client('logs')
+            logs = retry(
+                client.describe_log_groups,
+                logGroupNamePrefix=self.log_group)['logGroups']
+            if [l for l in logs if l['logGroupName'] == self.log_group]:
+                retry(client.create_log_group,
+                      logGroupName=self.log_group)
         except ClientError as e:
             if Error.code(e) != Error.ResourceExists:
                 raise

{"golden_diff": "diff --git a/c7n/log.py b/c7n/log.py\n--- a/c7n/log.py\n+++ b/c7n/log.py\n@@ -37,6 +37,7 @@\n except ImportError:\n     import queue as Queue\n \n+from c7n.utils import get_retry\n \n FLUSH_MARKER = object()\n SHUTDOWN_MARKER = object()\n@@ -82,9 +83,15 @@\n         # cleanup atexit, custodian is a bit more explicitly scoping shutdown to\n         # each policy, so use a sentinel value to avoid deadlocks.\n         self.shutdown = False\n+        retry = get_retry(('ThrottlingException',))\n         try:\n-            self.session_factory().client(\n-                'logs').create_log_group(logGroupName=self.log_group)\n+            client = self.session_factory().client('logs')\n+            logs = retry(\n+                client.describe_log_groups,\n+                logGroupNamePrefix=self.log_group)['logGroups']\n+            if [l for l in logs if l['logGroupName'] == self.log_group]:\n+                retry(client.create_log_group,\n+                      logGroupName=self.log_group)\n         except ClientError as e:\n             if Error.code(e) != Error.ResourceExists:\n                 raise\n", "issue": "handle throttles on log group output\n\n", "before_files": [{"content": "# Copyright 2016 Capital One Services, LLC\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"\nPython Standard Logging integration with CloudWatch Logs\n\nDouble Buffered with background thread delivery.\n\nWe do an initial buffering on the log handler directly, to avoid\nsome of the overhead of pushing to the queue (albeit dubious as\nstd logging does default lock acquisition around handler emit).\nalso uses a single thread for all outbound. Background thread\nuses a separate session.\n\"\"\"\n\nimport boto3\nfrom botocore.exceptions import ClientError\n\nimport itertools\nimport logging\nfrom operator import itemgetter\nimport threading\nimport time\n\ntry:\n    import Queue\nexcept ImportError:\n    import queue as Queue\n\n\nFLUSH_MARKER = object()\nSHUTDOWN_MARKER = object()\n\nEMPTY = Queue.Empty\n\n\nclass Error(object):\n\n    AlreadyAccepted = \"DataAlreadyAcceptedException\"\n    InvalidToken = \"InvalidSequenceTokenException\"\n    ResourceExists = \"ResourceAlreadyExistsException\"\n\n    @staticmethod\n    def code(e):\n        return e.response.get('Error', {}).get('Code')\n\n\nclass CloudWatchLogHandler(logging.Handler):\n    \"\"\"Python Log Handler to Send to Cloud Watch Logs\n\n    http://goo.gl/eZGAEK\n    \"\"\"\n\n    batch_size = 20\n    batch_interval = 40\n    batch_min_buffer = 10\n\n    def __init__(self, log_group=__name__, log_stream=None,\n                 session_factory=None):\n        super(CloudWatchLogHandler, self).__init__()\n        self.log_group = log_group\n        self.log_stream = log_stream\n        self.session_factory = session_factory or boto3.Session\n        self.transport = None\n        self.queue = Queue.Queue()\n        self.threads = []\n        # do some basic buffering before sending to transport to minimize\n        # queue/threading overhead\n        self.buf = []\n        self.last_seen = time.time()\n        # Logging module internally is tracking all handlers, for final\n        # cleanup atexit, custodian is a bit more explicitly scoping shutdown to\n        # each policy, so use a sentinel value to avoid deadlocks.\n        self.shutdown = False\n        try:\n            self.session_factory().client(\n                'logs').create_log_group(logGroupName=self.log_group)\n        except ClientError as e:\n            if Error.code(e) != Error.ResourceExists:\n                raise\n\n    # Begin logging.Handler API\n    def emit(self, message):\n        \"\"\"Send logs\"\"\"\n        # We're sending messages asynchronously, bubble to caller when\n        # we've detected an error on the message. This isn't great,\n        # but options once we've gone async without a deferred/promise\n        # aren't great.\n        if self.transport and self.transport.error:\n            raise self.transport.error\n\n        # Sanity safety, people do like to recurse by attaching to\n        # root log :-(\n        if message.name.startswith('boto'):\n            return\n\n        msg = self.format_message(message)\n        if not self.transport:\n            self.start_transports()\n        self.buf.append(msg)\n        self.flush_buffers(\n            (message.created - self.last_seen >= self.batch_interval))\n\n        self.last_seen = message.created\n\n    def flush(self):\n        \"\"\"Ensure all logging output has been flushed.\"\"\"\n        if self.shutdown:\n            return\n        self.flush_buffers(force=True)\n        self.queue.put(FLUSH_MARKER)\n        self.queue.join()\n\n    def close(self):\n        if self.shutdown:\n            return\n        self.shutdown = True\n        self.queue.put(SHUTDOWN_MARKER)\n        self.queue.join()\n        for t in self.threads:\n            t.join()\n        self.threads = []\n\n    # End logging.Handler API\n\n    def format_message(self, msg):\n        \"\"\"format message.\"\"\"\n        return {'timestamp': int(msg.created * 1000),\n                'message': self.format(msg),\n                'stream': self.log_stream or msg.name,\n                'group': self.log_group}\n\n    def start_transports(self):\n        \"\"\"start thread transports.\"\"\"\n        self.transport = Transport(\n            self.queue, self.batch_size, self.batch_interval,\n            self.session_factory)\n        thread = threading.Thread(target=self.transport.loop)\n        self.threads.append(thread)\n        thread.daemon = True\n        thread.start()\n\n    def flush_buffers(self, force=False):\n        if not force and len(self.buf) < self.batch_min_buffer:\n            return\n        self.queue.put(self.buf)\n        self.buf = []\n\n\nclass Transport(object):\n\n    def __init__(self, queue, batch_size, batch_interval, session_factory):\n        self.queue = queue\n        self.batch_size = batch_size\n        self.batch_interval = batch_interval\n        self.client = session_factory().client('logs')\n        self.sequences = {}\n        self.buffers = {}\n        self.error = None\n\n    def create_stream(self, group, stream):\n        try:\n            self.client.create_log_stream(\n                logGroupName=group, logStreamName=stream)\n        except ClientError as e:\n            if Error.code(e) != Error.ResourceExists:\n                self.error = e\n                return False\n        return True\n\n    def send(self):\n        for k, messages in self.buffers.items():\n            self.send_group(k, messages)\n        self.buffers = {}\n\n    def send_group(self, k, messages):\n        group, stream = k.split('=', 1)\n        if stream not in self.sequences:\n            if not self.create_stream(group, stream):\n                return\n            self.sequences[stream] = None\n        params = dict(\n            logGroupName=group, logStreamName=stream,\n            logEvents=sorted(\n                messages, key=itemgetter('timestamp'), reverse=False))\n        if self.sequences[stream]:\n            params['sequenceToken'] = self.sequences[stream]\n        try:\n            response = self.client.put_log_events(**params)\n        except ClientError as e:\n            if Error.code(e) in (Error.AlreadyAccepted, Error.InvalidToken):\n                self.sequences[stream] = e.response['Error']['Message'].rsplit(\n                    \" \", 1)[-1]\n                return self.send_group(k, messages)\n            self.error = e\n            return\n        self.sequences[stream] = response['nextSequenceToken']\n\n    def loop(self):\n        def keyed(datum):\n            return \"%s=%s\" % (\n                datum.pop('group'), datum.pop('stream'))\n\n        while True:\n            try:\n                datum = self.queue.get(block=True, timeout=self.batch_interval)\n            except EMPTY:\n                if Queue is None:\n                    return\n                datum = None\n            if datum is None:\n                # Timeout reached, flush\n                self.send()\n                continue\n            elif datum == FLUSH_MARKER:\n                self.send()\n            elif datum == SHUTDOWN_MARKER:\n                self.queue.task_done()\n                return\n            else:\n                for k, group in itertools.groupby(datum, keyed):\n                    self.buffers.setdefault(k, []).extend(group)\n            self.queue.task_done()\n", "path": "c7n/log.py"}], "after_files": [{"content": "# Copyright 2016 Capital One Services, LLC\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"\nPython Standard Logging integration with CloudWatch Logs\n\nDouble Buffered with background thread delivery.\n\nWe do an initial buffering on the log handler directly, to avoid\nsome of the overhead of pushing to the queue (albeit dubious as\nstd logging does default lock acquisition around handler emit).\nalso uses a single thread for all outbound. Background thread\nuses a separate session.\n\"\"\"\n\nimport boto3\nfrom botocore.exceptions import ClientError\n\nimport itertools\nimport logging\nfrom operator import itemgetter\nimport threading\nimport time\n\ntry:\n    import Queue\nexcept ImportError:\n    import queue as Queue\n\nfrom c7n.utils import get_retry\n\nFLUSH_MARKER = object()\nSHUTDOWN_MARKER = object()\n\nEMPTY = Queue.Empty\n\n\nclass Error(object):\n\n    AlreadyAccepted = \"DataAlreadyAcceptedException\"\n    InvalidToken = \"InvalidSequenceTokenException\"\n    ResourceExists = \"ResourceAlreadyExistsException\"\n\n    @staticmethod\n    def code(e):\n        return e.response.get('Error', {}).get('Code')\n\n\nclass CloudWatchLogHandler(logging.Handler):\n    \"\"\"Python Log Handler to Send to Cloud Watch Logs\n\n    http://goo.gl/eZGAEK\n    \"\"\"\n\n    batch_size = 20\n    batch_interval = 40\n    batch_min_buffer = 10\n\n    def __init__(self, log_group=__name__, log_stream=None,\n                 session_factory=None):\n        super(CloudWatchLogHandler, self).__init__()\n        self.log_group = log_group\n        self.log_stream = log_stream\n        self.session_factory = session_factory or boto3.Session\n        self.transport = None\n        self.queue = Queue.Queue()\n        self.threads = []\n        # do some basic buffering before sending to transport to minimize\n        # queue/threading overhead\n        self.buf = []\n        self.last_seen = time.time()\n        # Logging module internally is tracking all handlers, for final\n        # cleanup atexit, custodian is a bit more explicitly scoping shutdown to\n        # each policy, so use a sentinel value to avoid deadlocks.\n        self.shutdown = False\n        retry = get_retry(('ThrottlingException',))\n        try:\n            client = self.session_factory().client('logs')\n            logs = retry(\n                client.describe_log_groups,\n                logGroupNamePrefix=self.log_group)['logGroups']\n            if [l for l in logs if l['logGroupName'] == self.log_group]:\n                retry(client.create_log_group,\n                      logGroupName=self.log_group)\n        except ClientError as e:\n            if Error.code(e) != Error.ResourceExists:\n                raise\n\n    # Begin logging.Handler API\n    def emit(self, message):\n        \"\"\"Send logs\"\"\"\n        # We're sending messages asynchronously, bubble to caller when\n        # we've detected an error on the message. This isn't great,\n        # but options once we've gone async without a deferred/promise\n        # aren't great.\n        if self.transport and self.transport.error:\n            raise self.transport.error\n\n        # Sanity safety, people do like to recurse by attaching to\n        # root log :-(\n        if message.name.startswith('boto'):\n            return\n\n        msg = self.format_message(message)\n        if not self.transport:\n            self.start_transports()\n        self.buf.append(msg)\n        self.flush_buffers(\n            (message.created - self.last_seen >= self.batch_interval))\n\n        self.last_seen = message.created\n\n    def flush(self):\n        \"\"\"Ensure all logging output has been flushed.\"\"\"\n        if self.shutdown:\n            return\n        self.flush_buffers(force=True)\n        self.queue.put(FLUSH_MARKER)\n        self.queue.join()\n\n    def close(self):\n        if self.shutdown:\n            return\n        self.shutdown = True\n        self.queue.put(SHUTDOWN_MARKER)\n        self.queue.join()\n        for t in self.threads:\n            t.join()\n        self.threads = []\n\n    # End logging.Handler API\n\n    def format_message(self, msg):\n        \"\"\"format message.\"\"\"\n        return {'timestamp': int(msg.created * 1000),\n                'message': self.format(msg),\n                'stream': self.log_stream or msg.name,\n                'group': self.log_group}\n\n    def start_transports(self):\n        \"\"\"start thread transports.\"\"\"\n        self.transport = Transport(\n            self.queue, self.batch_size, self.batch_interval,\n            self.session_factory)\n        thread = threading.Thread(target=self.transport.loop)\n        self.threads.append(thread)\n        thread.daemon = True\n        thread.start()\n\n    def flush_buffers(self, force=False):\n        if not force and len(self.buf) < self.batch_min_buffer:\n            return\n        self.queue.put(self.buf)\n        self.buf = []\n\n\nclass Transport(object):\n\n    def __init__(self, queue, batch_size, batch_interval, session_factory):\n        self.queue = queue\n        self.batch_size = batch_size\n        self.batch_interval = batch_interval\n        self.client = session_factory().client('logs')\n        self.sequences = {}\n        self.buffers = {}\n        self.error = None\n\n    def create_stream(self, group, stream):\n        try:\n            self.client.create_log_stream(\n                logGroupName=group, logStreamName=stream)\n        except ClientError as e:\n            if Error.code(e) != Error.ResourceExists:\n                self.error = e\n                return False\n        return True\n\n    def send(self):\n        for k, messages in self.buffers.items():\n            self.send_group(k, messages)\n        self.buffers = {}\n\n    def send_group(self, k, messages):\n        group, stream = k.split('=', 1)\n        if stream not in self.sequences:\n            if not self.create_stream(group, stream):\n                return\n            self.sequences[stream] = None\n        params = dict(\n            logGroupName=group, logStreamName=stream,\n            logEvents=sorted(\n                messages, key=itemgetter('timestamp'), reverse=False))\n        if self.sequences[stream]:\n            params['sequenceToken'] = self.sequences[stream]\n        try:\n            response = self.client.put_log_events(**params)\n        except ClientError as e:\n            if Error.code(e) in (Error.AlreadyAccepted, Error.InvalidToken):\n                self.sequences[stream] = e.response['Error']['Message'].rsplit(\n                    \" \", 1)[-1]\n                return self.send_group(k, messages)\n            self.error = e\n            return\n        self.sequences[stream] = response['nextSequenceToken']\n\n    def loop(self):\n        def keyed(datum):\n            return \"%s=%s\" % (\n                datum.pop('group'), datum.pop('stream'))\n\n        while True:\n            try:\n                datum = self.queue.get(block=True, timeout=self.batch_interval)\n            except EMPTY:\n                if Queue is None:\n                    return\n                datum = None\n            if datum is None:\n                # Timeout reached, flush\n                self.send()\n                continue\n            elif datum == FLUSH_MARKER:\n                self.send()\n            elif datum == SHUTDOWN_MARKER:\n                self.queue.task_done()\n                return\n            else:\n                for k, group in itertools.groupby(datum, keyed):\n                    self.buffers.setdefault(k, []).extend(group)\n            self.queue.task_done()\n", "path": "c7n/log.py"}]}